bws_rs/http/
axum.rs

1use axum::response::IntoResponse;
2use futures::StreamExt;
3use tokio::io::AsyncWriteExt;
4
5use crate::service::s3::VRequest;
6
7pub struct Request {
8    request: axum::http::Request<axum::body::Body>,
9    query: Option<std::collections::HashMap<String, String>>,
10}
11impl From<Request> for axum::extract::Request {
12    fn from(val: Request) -> Self {
13        val.request
14    }
15}
16impl From<axum::extract::Request> for Request {
17    fn from(value: axum::extract::Request) -> Self {
18        let query = value.uri().query().map(|query| {
19            query
20                .split("&")
21                .map(|item| {
22                    item.find("=")
23                        .map_or((item.to_string(), "".to_string()), |pos| {
24                            (item[..pos].to_string(), item[pos + 1..].to_string())
25                        })
26                })
27                .collect::<std::collections::HashMap<String, String>>()
28        });
29        Self {
30            request: value,
31            query,
32        }
33    }
34}
35impl crate::authorization::v4::VHeader for Request {
36    fn get_header(&self, key: &str) -> Option<String> {
37        self.request
38            .headers()
39            .get(key)
40            .and_then(|value| value.to_str().ok().map(|value| value.to_string()))
41    }
42
43    fn set_header(&mut self, key: &str, val: &str) {
44        let key: axum::http::HeaderName = key.to_string().parse().unwrap();
45        self.request.headers_mut().insert(key, val.parse().unwrap());
46    }
47
48    fn delete_header(&mut self, key: &str) {
49        self.request.headers_mut().remove(key);
50    }
51
52    fn rng_header(&self, mut cb: impl FnMut(&str, &str) -> bool) {
53        for (k, v) in self.request.headers().iter() {
54            if !cb(k.as_str(), unsafe {
55                std::str::from_utf8_unchecked(v.as_bytes())
56            }) {
57                return;
58            }
59        }
60    }
61}
62impl crate::service::s3::VRequest for Request {
63    fn method(&self) -> String {
64        self.request.method().as_str().to_string()
65    }
66
67    fn url_path(&self) -> String {
68        self.request.uri().path().to_string()
69    }
70
71    fn get_query(&self, k: &str) -> Option<String> {
72        self.query
73            .as_ref()
74            .and_then(|query| query.get(k))
75            .map(|v| v.to_string())
76    }
77
78    fn all_query(&self, mut cb: impl FnMut(&str, &str) -> bool) {
79        self.query
80            .as_ref()
81            .map(|query| query.iter().all(|(k, v)| cb(k, v)));
82    }
83}
84impl crate::service::s3::VRequestPlus for Request {
85    fn body<'a>(
86        self,
87    ) -> std::pin::Pin<
88        Box<dyn 'a + Send + std::future::Future<Output = Result<Vec<u8>, std::io::Error>>>,
89    > {
90        Box::pin(async move {
91            let mut bodystream = self.request.into_body().into_data_stream();
92            let mut ret = Vec::new();
93            while let Some(bodystream) = bodystream.next().await {
94                let bytes = bodystream.map_err(std::io::Error::other)?;
95                ret.extend_from_slice(bytes.iter().as_slice());
96            }
97            Ok(ret)
98        })
99    }
100}
101pub struct BodyReader(axum_core::body::BodyDataStream);
102impl crate::utils::io::PollRead for BodyReader {
103    fn poll_read<'a>(
104        &'a mut self,
105    ) -> std::pin::Pin<
106        Box<dyn 'a + Send + std::future::Future<Output = Result<Option<Vec<u8>>, String>>>,
107    > {
108        Box::pin(async move {
109            let data = self.0.next().await;
110            match data {
111                Some(ret) => match ret {
112                    Ok(ret) => Ok(Some(ret.to_vec())),
113                    Err(err) => Err(err.to_string()),
114                },
115                None => Ok(None),
116            }
117        })
118    }
119}
120impl crate::service::s3::BodyReader for Request {
121    type BodyReader = BodyReader;
122
123    fn get_body_reader<'b>(
124        self,
125    ) -> std::pin::Pin<
126        Box<dyn 'b + Send + std::future::Future<Output = Result<Self::BodyReader, String>>>,
127    > {
128        Box::pin(async move {
129            let ret: axum::body::Body = self.request.into_body();
130            Ok(BodyReader(ret.into_data_stream()))
131        })
132    }
133}
134pub struct HeaderWarp(axum::http::HeaderMap);
135impl crate::authorization::v4::VHeader for HeaderWarp {
136    fn get_header(&self, key: &str) -> Option<String> {
137        self.0
138            .get(key)
139            .and_then(|value| value.to_str().ok().map(|value| value.to_string()))
140    }
141
142    fn set_header(&mut self, key: &str, val: &str) {
143        let key: axum::http::HeaderName = key.to_string().parse().unwrap();
144        self.0.insert(key, val.parse().unwrap());
145    }
146
147    fn delete_header(&mut self, key: &str) {
148        self.0.remove(key);
149    }
150
151    fn rng_header(&self, mut cb: impl FnMut(&str, &str) -> bool) {
152        for (k, v) in self.0.iter() {
153            if !cb(k.as_str(), unsafe {
154                std::str::from_utf8_unchecked(v.as_bytes())
155            }) {
156                return;
157            }
158        }
159    }
160}
161impl crate::service::s3::HeaderTaker for Request {
162    type Head = HeaderWarp;
163
164    fn take_header(&self) -> Self::Head {
165        HeaderWarp(self.request.headers().clone())
166    }
167}
168pub struct Response {
169    status: u16,
170    headers: axum::http::HeaderMap,
171    body: tokio::io::BufWriter<Vec<u8>>,
172}
173impl Default for Response {
174    fn default() -> Self {
175        Self {
176            status: Default::default(),
177            headers: Default::default(),
178            body: tokio::io::BufWriter::new(Default::default()),
179        }
180    }
181}
182impl From<Response> for axum::response::Response {
183    fn from(val: Response) -> Self {
184        let mut respbuilder = axum::response::Response::builder().status(if val.status == 0 {
185            200
186        } else {
187            val.status
188        });
189        if !val.headers.is_empty() {
190            if let Some(header) = respbuilder.headers_mut() {
191                *header = val.headers;
192            }
193        }
194        let raw = val.body.into_inner();
195        log::info!("result length {}", raw.len());
196        respbuilder.body(raw.into()).unwrap()
197        // (
198        //     axum::http::StatusCode::from_u16(if val.status == 0 { 200 } else { val.status })
199        //         .unwrap(),
200        //     raw,
201        // )
202        //     .into_response()
203    }
204}
205impl crate::authorization::v4::VHeader for Response {
206    fn get_header(&self, key: &str) -> Option<String> {
207        self.headers
208            .get(key)
209            .and_then(|v| v.to_str().ok().map(|v| v.to_string()))
210    }
211
212    fn set_header(&mut self, key: &str, val: &str) {
213        log::info!("set header {key} {val}");
214        self.headers.insert(
215            key.to_string().parse::<axum::http::HeaderName>().unwrap(),
216            val.parse().unwrap(),
217        );
218    }
219
220    fn delete_header(&mut self, key: &str) {
221        self.headers.remove(key);
222    }
223
224    fn rng_header(&self, mut cb: impl FnMut(&str, &str) -> bool) {
225        self.headers.iter().all(|(k, v)| {
226            cb(k.as_str(), unsafe {
227                std::str::from_utf8_unchecked(v.as_bytes())
228            })
229        });
230    }
231}
232pub struct BodyWriter<'a>(&'a mut tokio::io::BufWriter<Vec<u8>>);
233impl<'b> crate::utils::io::PollWrite for BodyWriter<'b> {
234    fn poll_write<'a>(
235        &'a mut self,
236        buff: &'a [u8],
237    ) -> std::pin::Pin<
238        Box<dyn 'a + Send + std::future::Future<Output = Result<usize, std::io::Error>>>,
239    > {
240        Box::pin(async move {
241            log::info!("write buff {}", buff.len());
242            let _ = self.0.write_all(buff).await;
243            let _ = self.0.flush().await;
244            Ok(buff.len())
245        })
246    }
247}
248impl crate::service::s3::BodyWriter for Response {
249    type BodyWriter<'a>
250    = BodyWriter<'a> where Self: 'a;
251
252    fn get_body_writer<'b>(
253        &'b mut self,
254    ) -> std::pin::Pin<
255        Box<dyn 'b + Send + std::future::Future<Output = Result<Self::BodyWriter<'_>, String>>>,
256    > {
257        Box::pin(async move { Ok(BodyWriter(&mut self.body)) })
258    }
259}
260impl crate::service::s3::VResponse for Response {
261    fn set_status(&mut self, status: u16) {
262        self.status = status;
263    }
264
265    fn send_header(&mut self) {}
266}
267pub async fn handle_fn(
268    req: axum::extract::Request<axum::body::Body>,
269    _next: axum::middleware::Next,
270) -> axum::response::Response {
271    use crate::service::s3::*;
272    use axum::http::StatusCode;
273    use std::sync::Arc;
274    let multipart_obj = req
275        .extensions()
276        .get::<Arc<dyn MultiUploadObjectHandler + Send + Sync>>()
277        .cloned();
278    match *req.method() {
279        axum::http::Method::PUT => {
280            let put_obj = req
281                .extensions()
282                .get::<std::sync::Arc<dyn crate::service::s3::PutObjectHandler + Sync + Send>>()
283                .cloned();
284            let create_bkt_obj = req
285                .extensions()
286                .get::<std::sync::Arc<dyn crate::service::s3::CreateBucketHandler + Sync + Send>>()
287                .cloned();
288            let v4head = req
289                .extensions()
290                .get::<crate::authorization::v4::V4Head>()
291                .cloned();
292            let path = req.uri().path();
293            let rpath = path
294                .trim_start_matches('/')
295                .splitn(2, '/')
296                .collect::<Vec<&str>>();
297            let rpath_len = rpath.len();
298            if rpath_len == 0 {
299                log::info!("args length invalid");
300                (StatusCode::BAD_REQUEST, b"").into_response()
301            } else {
302                let is_create_bkt = rpath_len == 1 || (rpath_len == 2 && rpath[1].is_empty());
303                let req = Request::from(req);
304                let mut resp = Response::default();
305                if is_create_bkt {
306                    //create bucket
307                    match create_bkt_obj {
308                        Some(create_bkt_obj) => {
309                            crate::service::s3::handle_create_bucket(
310                                req,
311                                &mut resp,
312                                &create_bkt_obj,
313                            )
314                            .await;
315                        }
316                        None => {
317                            log::warn!("not open create bucket method");
318                            return (StatusCode::FORBIDDEN, b"").into_response();
319                        }
320                    }
321                } else {
322                    let xid = req.get_query("x-id");
323                    if let Some(xid) = xid {
324                        if xid.as_str() == "UploadPart" {
325                            let mut resp = Response::default();
326
327                            // let upload_id = req.get_query("uploadId");
328                            // let part_number = req.get_query("partNumber");
329                            // if upload_id.is_none() || part_number.is_none() {
330                            //     return (axum::http::StatusCode::BAD_REQUEST, b"").into_response();
331                            // }
332
333                            return match multipart_obj {
334                                Some(multipart_obj) => {
335                                    handle_multipart_upload_part(req, &mut resp, &multipart_obj)
336                                        .await;
337                                    resp.into()
338                                }
339                                None => (axum::http::StatusCode::INTERNAL_SERVER_ERROR, b"")
340                                    .into_response(),
341                            };
342                        }
343                    }
344
345                    //put object
346                    match put_obj {
347                        Some(put_obj) => {
348                            crate::service::s3::handle_put_object(
349                                v4head.unwrap(),
350                                req,
351                                &mut resp,
352                                &put_obj,
353                            )
354                            .await;
355                        }
356                        None => {
357                            log::warn!("not open put object method");
358                            return (StatusCode::FORBIDDEN, b"").into_response();
359                        }
360                    }
361                }
362                resp.into()
363            }
364        }
365        axum::http::Method::GET => {
366            if req.uri().path().starts_with("/probe-bsign") {
367                return (axum::http::StatusCode::OK, b"").into_response();
368            }
369            let get_obj = req
370                .extensions()
371                .get::<Arc<dyn crate::service::s3::GetObjectHandler + Send + Sync>>()
372                .cloned();
373            let listbkt_obj = req
374                .extensions()
375                .get::<Arc<dyn crate::service::s3::ListBucketHandler + Send + Sync>>()
376                .cloned();
377            let getbkt_loc_obj = req
378                .extensions()
379                .get::<Arc<dyn crate::service::s3::GetBucketLocationHandler + Send + Sync>>()
380                .cloned();
381            let req = Request::from(req);
382            let url_path = req.url_path();
383            log::info!("path is {}", url_path.trim_start_matches('/').is_empty());
384            if let Some(lt) = req.get_query("list-type") {
385                if lt == "2" || url_path.trim_start_matches('/').is_empty() {
386                    log::info!("is list bucket");
387                    //get bucket object
388                    match listbkt_obj {
389                        Some(listbkt_obj) => {
390                            let mut resp = Response::default();
391                            crate::service::s3::handle_get_list_buckets(
392                                req,
393                                &mut resp,
394                                &listbkt_obj,
395                            )
396                            .await;
397                            return resp.into();
398                        }
399                        None => {
400                            log::warn!("not open head method");
401                            let ret = (StatusCode::FORBIDDEN, b"").into_response();
402                            return ret;
403                        }
404                    }
405                }
406            } else if url_path.trim_start_matches('/').is_empty() {
407                log::info!("is list bucket");
408                //get bucket object
409                match listbkt_obj {
410                    Some(listbkt_obj) => {
411                        let mut resp = Response::default();
412                        crate::service::s3::handle_get_list_buckets(req, &mut resp, &listbkt_obj)
413                            .await;
414                        return resp.into();
415                    }
416                    None => {
417                        log::warn!("not open head method");
418                        let ret = (StatusCode::FORBIDDEN, b"").into_response();
419                        return ret;
420                    }
421                }
422            }
423            if let Some(loc) = req.get_query("location") {
424                //get bucket location
425                return match getbkt_loc_obj {
426                    Some(bkt) => {
427                        match bkt
428                            .handle(if loc.is_empty() { None } else { Some(&loc) })
429                            .await
430                        {
431                            Ok(loc) => {
432                                let lc = match loc {
433                                    Some(loc) => bucket::LocationConstraint::new(loc),
434                                    None => bucket::LocationConstraint::new(""),
435                                };
436                                match quick_xml::se::to_string(&lc) {
437                                    Ok(content) => (StatusCode::OK, content).into_response(),
438                                    Err(err) => {
439                                        log::error!("xml encode error {err}");
440                                        (StatusCode::INTERNAL_SERVER_ERROR, b"").into_response()
441                                    }
442                                }
443                            }
444                            Err(_) => {
445                                log::error!("get bucket location error");
446                                (StatusCode::INTERNAL_SERVER_ERROR, b"").into_response()
447                            }
448                        }
449                    }
450                    None => {
451                        log::warn!("not open get bucket location method");
452                        (StatusCode::FORBIDDEN, b"").into_response()
453                    }
454                };
455            }
456            //get object
457            match get_obj {
458                Some(obj) => {
459                    let mut resp = Response::default();
460                    crate::service::s3::handle_get_object(req, &mut resp, &obj).await;
461                    resp.into()
462                }
463                None => {
464                    log::warn!("not open get object method");
465                    (StatusCode::FORBIDDEN, b"").into_response()
466                }
467            }
468        }
469        axum::http::Method::DELETE => {
470            let path = req.uri().path().trim_start_matches('/');
471            if path.is_empty() {
472                return (StatusCode::BAD_REQUEST, b"").into_response();
473            }
474            let rr = path.split("/").collect::<Vec<&str>>();
475            let rr_len = rr.len();
476            if rr_len == 1 || (rr_len == 2 && rr[1].is_empty()) {
477                match req
478                    .extensions()
479                    .get::<Arc<dyn DeleteBucketHandler + Send + Sync>>()
480                    .cloned()
481                {
482                    Some(delete_bkt_obj) => {
483                        let mut resp = Response::default();
484                        handle_delete_bucket(Request::from(req), &mut resp, &delete_bkt_obj).await;
485                        resp.into()
486                    }
487                    None => {
488                        log::warn!("not open get delete bucket method");
489                        (StatusCode::FORBIDDEN, b"").into_response()
490                    }
491                }
492            } else {
493                match req
494                    .extensions()
495                    .get::<Arc<dyn DeleteObjectHandler + Send + Sync>>()
496                    .cloned()
497                {
498                    Some(delete_obj_obj) => {
499                        let mut resp = Response::default();
500                        handle_delete_object(Request::from(req), &mut resp, &delete_obj_obj).await;
501                        resp.into()
502                    }
503                    None => {
504                        log::warn!("not open get delete bucket method");
505                        (StatusCode::FORBIDDEN, b"").into_response()
506                    }
507                }
508            }
509        }
510        axum::http::Method::HEAD => {
511            let head_obj = req
512                .extensions()
513                .get::<std::sync::Arc<dyn crate::service::s3::HeadHandler + Sync + Send>>()
514                .cloned();
515            if head_obj.is_none() {
516                log::warn!("not open head features");
517                return (StatusCode::INTERNAL_SERVER_ERROR, b"").into_response();
518            }
519            let head_obj = head_obj.unwrap();
520            let req = Request::from(req);
521            let raw_path = req.url_path();
522            let args = raw_path
523                .trim_start_matches('/')
524                .splitn(2, '/')
525                .collect::<Vec<&str>>();
526            if args.len() != 2 {
527                return (StatusCode::BAD_REQUEST, b"").into_response();
528            }
529            match head_obj.lookup(args[0], args[1]).await {
530                Ok(metadata) => match metadata {
531                    Some(head) => {
532                        use crate::authorization::v4::VHeader;
533                        let mut resp = Response::default();
534                        if let Some(v) = head.content_length {
535                            resp.set_header("content-length", v.to_string().as_str())
536                        }
537                        if let Some(v) = head.etag {
538                            resp.set_header("etag", &v);
539                        }
540                        if let Some(v) = head.content_type {
541                            resp.set_header("content-type", &v);
542                        }
543                        if let Some(v) = head.last_modified {
544                            resp.set_header("last-modified", &v);
545                        }
546                        (StatusCode::OK, b"").into_response()
547                    }
548                    None => (StatusCode::NOT_FOUND, b"").into_response(),
549                },
550                Err(err) => {
551                    log::error!("lookup object metadata error {err}");
552                    (StatusCode::INTERNAL_SERVER_ERROR, b"").into_response()
553                }
554            }
555        }
556        axum::http::Method::POST => match multipart_obj {
557            Some(multipart_obj) => {
558                let mut resp = Response::default();
559                let is_create_session = if let Some(query) = req.uri().query() {
560                    query.contains("uploads=")
561                } else {
562                    false
563                };
564                let req = Request::from(req);
565                if is_create_session {
566                    handle_multipart_create_session(req, &mut resp, &multipart_obj).await;
567                } else if req.get_query("uploadId").is_some() {
568                    handle_multipart_complete_session(req, &mut resp, &multipart_obj).await;
569                } else {
570                    return (StatusCode::BAD_REQUEST, b"").into_response();
571                }
572                resp.into()
573            }
574            None => {
575                log::warn!("not open multipart object features");
576                (StatusCode::INTERNAL_SERVER_ERROR, b"").into_response()
577            }
578        },
579        _ => (StatusCode::METHOD_NOT_ALLOWED, b"").into_response(),
580    }
581}
582pub async fn handle_authorization_middleware(
583    req: axum::extract::Request<axum::body::Body>,
584    next: axum::middleware::Next,
585) -> impl axum::response::IntoResponse {
586    let ret = req
587        .extensions()
588        .get::<std::sync::Arc<dyn crate::authorization::AccesskeyStore + Send + Sync>>()
589        .cloned();
590    let ak_store = match ret {
591        Some(ret) => ret,
592        None => {
593            return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, b"").into_response();
594        }
595    };
596    let req = Request::from(req);
597    let base_arg = match crate::authorization::v4::extract_args(&req) {
598        Ok(arg) => arg,
599        Err(_) => {
600            return (axum::http::StatusCode::BAD_REQUEST, b"").into_response();
601        }
602    };
603    let mut query = Vec::new();
604    req.all_query(|k, v| {
605        query.push(crate::utils::BaseKv {
606            key: k.to_string(),
607            val: v.to_string(),
608        });
609        true
610    });
611    let secretkey = match ak_store.get(&base_arg.access_key).await {
612        Ok(secretkey) => {
613            if secretkey.is_none() {
614                return (axum::http::StatusCode::FORBIDDEN, b"").into_response();
615            }
616            secretkey.unwrap()
617        }
618        Err(_) => {
619            return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, b"").into_response();
620        }
621    };
622    let ret = crate::authorization::v4::get_v4_signature(
623        &req,
624        req.method().as_str(),
625        &base_arg.region,
626        &base_arg.service,
627        req.url_path().as_str(),
628        &secretkey,
629        &base_arg.content_hash,
630        &base_arg.signed_headers,
631        query,
632    );
633    let circle_hasher = match ret {
634        Ok((sig, circle_hasher)) => {
635            if sig != base_arg.signature {
636                log::info!(
637                    "expect {sig} got {} args {:?} url_path {}",
638                    base_arg.signature,
639                    base_arg,
640                    req.url_path().as_str()
641                );
642                return (axum::http::StatusCode::FORBIDDEN, b"").into_response();
643            }
644            circle_hasher
645        }
646        Err(err) => {
647            log::error!("signature failed {err}");
648            return (axum::http::StatusCode::FORBIDDEN, b"").into_response();
649        }
650    };
651    let v4head = crate::authorization::v4::V4Head::new(
652        base_arg.signature,
653        base_arg.region,
654        base_arg.access_key,
655        circle_hasher,
656    );
657    let mut req: axum::http::Request<axum::body::Body> = req.into();
658    req.extensions_mut().insert(v4head);
659    next.run(req).await
660}
661mod bucket {
662
663    #[derive(serde::Serialize, Debug)]
664    #[serde(rename = "LocationConstraint", rename_all = "PascalCase")]
665    pub struct LocationConstraint {
666        #[serde(rename = "$value")]
667        region: String,
668
669        #[serde(rename = "xmlns")]
670        _xmlns: &'static str,
671    }
672    impl LocationConstraint {
673        pub fn new<T: Into<String>>(region: T) -> Self {
674            Self {
675                region: region.into(),
676                _xmlns: "http://s3.amazonaws.com/doc/2006-03-01/",
677            }
678        }
679    }
680}
681#[cfg(test)]
682mod itest {
683    use std::sync::Arc;
684
685    use tokio::io::AsyncReadExt;
686
687    #[derive(Default)]
688    struct Target {}
689    use crate::service::s3::*;
690    impl CreateBucketHandler for Target {
691        fn handle<'a>(
692            &'a self,
693            _opt: &'a CreateBucketOption,
694            _bucket: &'a str,
695        ) -> std::pin::Pin<Box<dyn 'a + Send + std::future::Future<Output = Result<(), String>>>>
696        {
697            Box::pin(async move {
698                log::info!("create bucket {_bucket}");
699                Ok(())
700            })
701        }
702    }
703    impl ListBucketHandler for Target {
704        fn handle<'a>(
705            &'a self,
706            _opt: &'a ListBucketsOption,
707        ) -> std::pin::Pin<
708            Box<dyn 'a + Send + std::future::Future<Output = Result<Vec<Bucket>, String>>>,
709        > {
710            Box::pin(async move {
711                let datetime = chrono::Utc::now().to_rfc3339();
712                Ok(vec![Bucket {
713                    name: "test1".to_string(),
714                    creation_date: datetime,
715                    bucket_region: "us-east-1".to_string(),
716                }])
717            })
718        }
719    }
720    impl HeadHandler for Target {
721        fn lookup<'a>(
722            &self,
723            _bucket: &str,
724            _object: &str,
725        ) -> std::pin::Pin<
726            Box<
727                dyn 'a
728                    + Send
729                    + Sync
730                    + std::future::Future<Output = Result<Option<HeadObjectResult>, Error>>,
731            >,
732        > {
733            Box::pin(async move {
734                let mut ret: HeadObjectResult = Default::default();
735                ret.checksum_sha256 = Some(
736                    "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824".to_string(),
737                );
738                ret.content_length = Some(5);
739                ret.etag = Some("5d41402abc4b2a76b9719d911017c592".to_string());
740                ret.last_modified = Some(
741                    chrono::Utc::now()
742                        .format("%a, %d %b %Y %H:%M:%S GMT")
743                        .to_string(),
744                );
745                Ok(Some(ret))
746            })
747        }
748    }
749    impl PutObjectHandler for Target {
750        fn handle<'a>(
751            &'a self,
752            opt: &PutObjectOption,
753            bucket: &'a str,
754            object: &'a str,
755            body: &'a mut (dyn tokio::io::AsyncRead + Unpin + Send),
756        ) -> std::pin::Pin<Box<dyn 'a + Send + std::future::Future<Output = Result<(), String>>>>
757        {
758            Box::pin(async move {
759                log::info!("put bucket {bucket} object {object}");
760                let mut buff = vec![];
761                match body.read_to_end(&mut buff).await {
762                    Ok(size) => {
763                        log::info!("get {}", unsafe {
764                            std::str::from_utf8_unchecked(&buff[..size])
765                        });
766                    }
767                    Err(err) => {
768                        log::error!("read error {err}");
769                    }
770                }
771                Ok(())
772            })
773        }
774    }
775    impl DeleteBucketHandler for Target {
776        fn handle<'a>(
777            &'a self,
778            _opt: &'a DeleteBucketOption,
779            _bucket: &'a str,
780        ) -> std::pin::Pin<Box<dyn 'a + Send + std::future::Future<Output = Result<(), String>>>>
781        {
782            Box::pin(async move {
783                log::info!("delete bucket {_bucket}");
784                Ok(())
785            })
786        }
787    }
788    impl DeleteObjectHandler for Target {
789        fn handle<'a>(
790            &'a self,
791            _opt: &'a DeleteObjectOption,
792            _object: &'a str,
793        ) -> std::pin::Pin<Box<dyn 'a + Send + std::future::Future<Output = Result<(), String>>>>
794        {
795            Box::pin(async move {
796                log::info!("delete object {_object}");
797                Ok(())
798            })
799        }
800    }
801    impl crate::authorization::AccesskeyStore for Target {
802        fn get<'a>(
803            &'a self,
804            _accesskey: &'a str,
805        ) -> std::pin::Pin<
806            Box<
807                dyn 'a + Send + Sync + std::future::Future<Output = Result<Option<String>, String>>,
808            >,
809        > {
810            Box::pin(async move { Ok(Some(format!("{_accesskey}12345"))) })
811        }
812    }
813    impl crate::service::s3::GetObjectHandler for Target {
814        fn handle<'a>(
815            &'a self,
816            bucket: &str,
817            object: &str,
818            opt: crate::service::s3::GetObjectOption,
819            mut out: tokio::sync::Mutex<
820                std::pin::Pin<
821                    std::boxed::Box<(dyn crate::utils::io::PollWrite + Send + Unpin + 'a)>,
822                >,
823            >,
824        ) -> std::pin::Pin<Box<dyn 'a + Send + std::future::Future<Output = Result<(), String>>>>
825        {
826            Box::pin(async move {
827                let mut l = out.lock().await;
828                let _ = l.poll_write(b"hello").await.map_err(|err| {
829                    log::error!("write error {err}");
830                });
831                Ok(())
832            })
833        }
834    }
835    impl crate::service::s3::GetBucketLocationHandler for Target {}
836    impl MultiUploadObjectHandler for Target {
837        fn handle_create_session<'a>(
838            &'a self,
839            bucket: &'a str,
840            key: &'a str,
841        ) -> std::pin::Pin<Box<dyn 'a + Send + std::future::Future<Output = Result<String, ()>>>>
842        {
843            Box::pin(async move { Ok("ffffff".to_string()) })
844        }
845
846        fn handle_upload_part<'a>(
847            &'a self,
848            bucket: &'a str,
849            key: &'a str,
850            upload_id: &'a str,
851            part_number: u32,
852            body: &'a mut (dyn tokio::io::AsyncRead + Unpin + Send),
853        ) -> std::pin::Pin<Box<dyn 'a + Send + std::future::Future<Output = Result<String, ()>>>>
854        {
855            Box::pin(async move {
856                let mut buff = Vec::new();
857                let size = body
858                    .read_to_end(&mut buff)
859                    .await
860                    .map_err(|err| log::error!("read body error {err}"))?;
861                println!(
862                    "upload part upload_id={upload_id} part_number={part_number} bucket={bucket} key={key}\n{}",
863                    unsafe { std::str::from_boxed_utf8_unchecked((&buff[..size]).into()) }
864                );
865                Ok("5d41402abc4b2a76b9719d911017c592".to_string())
866            })
867        }
868
869        fn handle_complete<'a>(
870            &'a self,
871            bucket: &'a str,
872            key: &'a str,
873            upload_id: &'a str,
874            //(etag,part number)
875            data: &'a [(&'a str, u32)],
876            opts: MultiUploadObjectCompleteOption,
877        ) -> std::pin::Pin<Box<dyn 'a + Send + std::future::Future<Output = Result<String, ()>>>>
878        {
879            Box::pin(async move { Ok("69a329523ce1ec88bf63061863d9cb14".to_string()) })
880        }
881
882        fn handle_abort<'a>(
883            &'a self,
884            bucket: &'a str,
885            key: &'a str,
886            upload_id: &'a str,
887        ) -> std::pin::Pin<Box<dyn 'a + Send + std::future::Future<Output = Result<(), ()>>>>
888        {
889            todo!()
890        }
891    }
892    #[tokio::test]
893    async fn test_server() -> Result<(), Box<dyn std::error::Error>> {
894        let _ = tokio::fs::create_dir_all(".sys_bws").await;
895        env_logger::builder()
896            .filter_level(log::LevelFilter::Info)
897            .init();
898        let target = Arc::new(Target::default());
899        let r = axum::Router::new()
900            .layer(axum::middleware::from_fn(super::handle_fn))
901            .layer(axum::middleware::from_fn(
902                super::handle_authorization_middleware,
903            ))
904            .layer(axum::Extension(
905                target.clone() as Arc<dyn PutObjectHandler + Send + Sync>
906            ))
907            .layer(axum::Extension(
908                target.clone() as Arc<dyn HeadHandler + Send + Sync>
909            ))
910            .layer(axum::Extension(
911                target.clone() as Arc<dyn ListBucketHandler + Send + Sync>
912            ))
913            .layer(axum::Extension(
914                target.clone() as Arc<dyn CreateBucketHandler + Send + Sync>
915            ))
916            .layer(axum::Extension(
917                target.clone() as Arc<dyn DeleteBucketHandler + Send + Sync>
918            ))
919            .layer(axum::Extension(
920                target.clone() as Arc<dyn DeleteObjectHandler + Send + Sync>
921            ))
922            .layer(axum::Extension(
923                target.clone() as Arc<dyn crate::authorization::AccesskeyStore + Send + Sync>
924            ))
925            .layer(axum::Extension(
926                target.clone() as Arc<dyn GetObjectHandler + Send + Sync>
927            ))
928            .layer(axum::Extension(
929                target.clone() as Arc<dyn GetBucketLocationHandler + Send + Sync>
930            )).layer(axum::Extension(
931                target.clone() as Arc<dyn MultiUploadObjectHandler + Send + Sync>
932            ));
933        let l = tokio::net::TcpListener::bind("0.0.0.0:9900").await?;
934        axum::serve(l, r).await?;
935        Ok(())
936    }
937}
bws_rs/http/axum.rs

bws_rs/http/
axum.rs
