use futures_core::ready;
use http::{Request, Response};
use http_body::Body;
use pin_project::pin_project;
use std::{
future::Future,
pin::Pin,
task::{Context, Poll},
};
use tower_layer::Layer;
use tower_service::Service;
#[derive(Debug, Clone)]
pub struct AsyncRequireAuthorizationLayer<T> {
auth: T,
}
impl<T> AsyncRequireAuthorizationLayer<T>
where
T: AsyncAuthorizeRequest,
{
pub fn new(auth: T) -> AsyncRequireAuthorizationLayer<T> {
Self { auth }
}
}
impl<S, T> Layer<S> for AsyncRequireAuthorizationLayer<T>
where
T: Clone + AsyncAuthorizeRequest,
{
type Service = AsyncRequireAuthorization<S, T>;
fn layer(&self, inner: S) -> Self::Service {
AsyncRequireAuthorization::new(inner, self.auth.clone())
}
}
#[derive(Clone, Debug)]
pub struct AsyncRequireAuthorization<S, T> {
inner: S,
auth: T,
}
impl<S, T> AsyncRequireAuthorization<S, T> {
define_inner_service_accessors!();
}
impl<S, T> AsyncRequireAuthorization<S, T>
where
T: AsyncAuthorizeRequest,
{
pub fn new(inner: S, auth: T) -> AsyncRequireAuthorization<S, T> {
Self { inner, auth }
}
}
impl<ReqBody, ResBody, S, T> Service<Request<ReqBody>> for AsyncRequireAuthorization<S, T>
where
S: Service<Request<ReqBody>, Response = Response<ResBody>> + Clone,
ResBody: Default,
T: AsyncAuthorizeRequest<ResponseBody = ResBody> + Clone,
{
type Response = Response<ResBody>;
type Error = S::Error;
type Future = ResponseFuture<T, S, ReqBody>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.inner.poll_ready(cx)
}
fn call(&mut self, req: Request<ReqBody>) -> Self::Future {
let auth = self.auth.clone();
let inner = self.inner.clone();
let authorize = self.auth.authorize(&req);
ResponseFuture {
auth,
state: State::Authorize {
authorize,
req: Some(req),
},
service: inner,
}
}
}
#[pin_project(project = StateProj)]
enum State<A, ReqBody, SFut> {
Authorize {
#[pin]
authorize: A,
req: Option<Request<ReqBody>>,
},
Authorized {
#[pin]
fut: SFut,
},
}
#[pin_project]
pub struct ResponseFuture<Auth, S, ReqBody>
where
Auth: AsyncAuthorizeRequest,
S: Service<Request<ReqBody>>,
{
auth: Auth,
#[pin]
state: State<Auth::Future, ReqBody, S::Future>,
service: S,
}
impl<Auth, S, ReqBody, B> Future for ResponseFuture<Auth, S, ReqBody>
where
Auth: AsyncAuthorizeRequest<ResponseBody = B>,
S: Service<Request<ReqBody>, Response = Response<B>>,
{
type Output = Result<Response<B>, S::Error>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let mut this = self.project();
loop {
match this.state.as_mut().project() {
StateProj::Authorize { authorize, req } => {
let auth = ready!(authorize.poll(cx));
let mut req = req.take().expect("future polled after completion");
match auth {
Some(output) => {
this.auth.on_authorized(&mut req, output);
let fut = this.service.call(req);
this.state.set(State::Authorized { fut })
}
None => {
let res = this.auth.unauthorized_response(&req);
return Poll::Ready(Ok(res));
}
};
}
StateProj::Authorized { fut } => {
return fut.poll(cx);
}
}
}
}
}
pub trait AsyncAuthorizeRequest {
type Output;
type Future: Future<Output = Option<Self::Output>>;
type ResponseBody: Body;
fn authorize<B>(&mut self, request: &Request<B>) -> Self::Future;
#[inline]
fn on_authorized<B>(&mut self, _request: &mut Request<B>, _output: Self::Output) {}
fn unauthorized_response<B>(&mut self, request: &Request<B>) -> Response<Self::ResponseBody>;
}
#[cfg(test)]
mod tests {
#[allow(unused_imports)]
use super::*;
use futures_util::future::BoxFuture;
use http::{header, StatusCode};
use hyper::Body;
use tower::{BoxError, ServiceBuilder, ServiceExt};
#[derive(Clone, Copy)]
struct MyAuth;
impl AsyncAuthorizeRequest for MyAuth {
type Output = UserId;
type Future = BoxFuture<'static, Option<UserId>>;
type ResponseBody = Body;
fn authorize<B>(&mut self, request: &Request<B>) -> Self::Future {
let authorized = request
.headers()
.get(header::AUTHORIZATION)
.and_then(|it| it.to_str().ok())
.and_then(|it| it.strip_prefix("Bearer "))
.map(|it| it == "69420")
.unwrap_or(false);
Box::pin(async move {
if authorized {
Some(UserId(String::from("6969")))
} else {
None
}
})
}
fn on_authorized<B>(&mut self, request: &mut Request<B>, user_id: UserId) {
request.extensions_mut().insert(user_id);
}
fn unauthorized_response<B>(&mut self, _request: &Request<B>) -> Response<Body> {
Response::builder()
.status(StatusCode::UNAUTHORIZED)
.body(Body::empty())
.unwrap()
}
}
#[derive(Debug)]
struct UserId(String);
#[tokio::test]
async fn require_async_auth_works() {
let mut service = ServiceBuilder::new()
.layer(AsyncRequireAuthorizationLayer::new(MyAuth))
.service_fn(echo);
let request = Request::get("/")
.header(header::AUTHORIZATION, "Bearer 69420")
.body(Body::empty())
.unwrap();
let res = service.ready().await.unwrap().call(request).await.unwrap();
assert_eq!(res.status(), StatusCode::OK);
}
#[tokio::test]
async fn require_async_auth_401() {
let mut service = ServiceBuilder::new()
.layer(AsyncRequireAuthorizationLayer::new(MyAuth))
.service_fn(echo);
let request = Request::get("/")
.header(header::AUTHORIZATION, "Bearer deez")
.body(Body::empty())
.unwrap();
let res = service.ready().await.unwrap().call(request).await.unwrap();
assert_eq!(res.status(), StatusCode::UNAUTHORIZED);
}
async fn echo(req: Request<Body>) -> Result<Response<Body>, BoxError> {
Ok(Response::new(req.into_body()))
}
}