use std::marker::PhantomData;
use std::task::{Context, Poll};
use axum::extract::Request;
use axum::response::{IntoResponse, Response};
use futures_util::{FutureExt, future::BoxFuture};
use tower::{Layer, Service};
use unitycatalog_common::{Error, Result};
use crate::api::RequestContext;
use crate::policy::Principal;
pub const DEFAULT_FORWARDED_USER_HEADER: &str = "x-forwarded-user";
impl<S: Send + Sync> axum::extract::FromRequestParts<S> for RequestContext {
type Rejection = std::convert::Infallible;
async fn from_request_parts(
parts: &mut axum::http::request::Parts,
_state: &S,
) -> std::result::Result<Self, Self::Rejection> {
let recipient = parts
.extensions
.get::<Principal>()
.cloned()
.unwrap_or_else(Principal::anonymous);
Ok(RequestContext { recipient })
}
}
pub trait Authenticator<I: Clone + Send + Sync + 'static>: Send + Sync + 'static {
fn authenticate(&self, request: &Request) -> Result<I>;
}
#[derive(Clone)]
pub struct AnonymousAuthenticator;
impl Authenticator<Principal> for AnonymousAuthenticator {
fn authenticate(&self, _: &Request) -> Result<Principal> {
Ok(Principal::anonymous())
}
}
#[derive(Clone)]
pub struct ReverseProxyAuthenticator {
header: String,
on_missing: OnMissingIdentity,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum OnMissingIdentity {
Reject,
Anonymous,
}
impl ReverseProxyAuthenticator {
pub fn new() -> Self {
Self {
header: DEFAULT_FORWARDED_USER_HEADER.to_string(),
on_missing: OnMissingIdentity::Reject,
}
}
pub fn with_header(mut self, header: impl Into<String>) -> Self {
self.header = header.into();
self
}
pub fn with_on_missing(mut self, on_missing: OnMissingIdentity) -> Self {
self.on_missing = on_missing;
self
}
}
impl Default for ReverseProxyAuthenticator {
fn default() -> Self {
Self::new()
}
}
impl Authenticator<Principal> for ReverseProxyAuthenticator {
fn authenticate(&self, request: &Request) -> Result<Principal> {
match request.headers().get(&self.header) {
Some(value) => {
let name = value.to_str().map_err(|_| {
Error::unauthenticated(format!("`{}` header is not valid UTF-8", self.header))
})?;
let name = name.trim();
if name.is_empty() {
return Err(Error::unauthenticated(format!(
"`{}` header is empty",
self.header
)));
}
Ok(Principal::user(name))
}
None => match self.on_missing {
OnMissingIdentity::Anonymous => Ok(Principal::anonymous()),
OnMissingIdentity::Reject => Err(Error::unauthenticated(format!(
"missing `{}` header",
self.header
))),
},
}
}
}
#[derive(Clone)]
pub struct AuthenticationMiddleware<S, T, I = Principal> {
inner: S,
authenticator: T,
_identity: PhantomData<I>,
}
#[allow(unused)]
impl<S, T, I> AuthenticationMiddleware<S, T, I> {
pub fn new(inner: S, authenticator: T) -> Self {
Self {
inner,
authenticator,
_identity: PhantomData,
}
}
pub fn layer(authenticator: T) -> AuthenticationLayer<T, I> {
AuthenticationLayer::new(authenticator)
}
}
impl<S, T, I> Service<Request> for AuthenticationMiddleware<S, T, I>
where
S: Service<Request, Response = Response> + Send + 'static,
S::Future: Send + 'static,
T: Authenticator<I>,
I: Clone + Send + Sync + 'static,
{
type Response = S::Response;
type Error = S::Error;
type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
self.inner.poll_ready(cx)
}
fn call(&mut self, mut req: Request) -> Self::Future {
match self.authenticator.authenticate(&req) {
Ok(identity) => {
req.extensions_mut().insert(identity);
self.inner.call(req).boxed()
}
Err(e) => async { Ok(crate::Error::from(e).into_response()) }.boxed(),
}
}
}
#[derive(Clone)]
pub struct AuthenticationLayer<T, I = Principal> {
authenticator: T,
_identity: PhantomData<I>,
}
impl<T, I> AuthenticationLayer<T, I> {
pub fn new(authenticator: T) -> Self {
Self {
authenticator,
_identity: PhantomData,
}
}
}
impl<S, T, I> Layer<S> for AuthenticationLayer<T, I>
where
T: Clone + Send + Sync + 'static,
I: Clone + Send + Sync + 'static,
{
type Service = AuthenticationMiddleware<S, T, I>;
fn layer(&self, inner: S) -> Self::Service {
AuthenticationMiddleware {
inner,
authenticator: self.authenticator.clone(),
_identity: PhantomData,
}
}
}
#[cfg(test)]
mod tests {
use axum::body::Body;
use axum::extract::Request;
use axum::http::{StatusCode, header};
use tower::{ServiceBuilder, ServiceExt};
use super::*;
async fn check_recipient(req: Request) -> Result<Response<Body>> {
assert!(matches!(
req.extensions().get::<Principal>(),
Some(Principal::Anonymous) | Some(Principal::User(_))
));
Ok(Response::new(req.into_body()))
}
#[tokio::test]
async fn test_authentication_middleware() {
let authenticator = AnonymousAuthenticator {};
let mut service = ServiceBuilder::new()
.layer(AuthenticationLayer::new(authenticator))
.service_fn(check_recipient);
let request = Request::get("/")
.header(header::AUTHORIZATION, "Bearer foo")
.body(Body::empty())
.unwrap();
let response = service.ready().await.unwrap().call(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
let request = Request::get("/").body(Body::empty()).unwrap();
let response = service.ready().await.unwrap().call(request).await.unwrap();
assert_eq!(response.status(), StatusCode::OK);
}
#[test]
fn reverse_proxy_extracts_forwarded_user() {
let auth = ReverseProxyAuthenticator::new();
let req = Request::get("/")
.header(DEFAULT_FORWARDED_USER_HEADER, "alice")
.body(Body::empty())
.unwrap();
assert_eq!(auth.authenticate(&req).unwrap(), Principal::user("alice"));
}
#[test]
fn reverse_proxy_honors_custom_header_and_trims() {
let auth = ReverseProxyAuthenticator::new().with_header("x-user");
let req = Request::get("/")
.header("x-user", " bob ")
.body(Body::empty())
.unwrap();
assert_eq!(auth.authenticate(&req).unwrap(), Principal::user("bob"));
}
#[test]
fn reverse_proxy_rejects_missing_header_by_default() {
let auth = ReverseProxyAuthenticator::new();
let req = Request::get("/").body(Body::empty()).unwrap();
assert!(auth.authenticate(&req).is_err());
}
#[test]
fn reverse_proxy_rejects_empty_header() {
let auth = ReverseProxyAuthenticator::new();
let req = Request::get("/")
.header(DEFAULT_FORWARDED_USER_HEADER, " ")
.body(Body::empty())
.unwrap();
assert!(auth.authenticate(&req).is_err());
}
#[test]
fn reverse_proxy_falls_back_to_anonymous_when_configured() {
let auth = ReverseProxyAuthenticator::new().with_on_missing(OnMissingIdentity::Anonymous);
let req = Request::get("/").body(Body::empty()).unwrap();
assert_eq!(auth.authenticate(&req).unwrap(), Principal::anonymous());
}
#[tokio::test]
async fn reverse_proxy_missing_identity_maps_to_401() {
let mut service = ServiceBuilder::new()
.layer(AuthenticationLayer::new(ReverseProxyAuthenticator::new()))
.service_fn(check_recipient);
let request = Request::get("/").body(Body::empty()).unwrap();
let response = service.ready().await.unwrap().call(request).await.unwrap();
assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
}
}