use std::{
convert::Infallible,
error::Error,
sync::Arc,
task::{Context, Poll},
};
use aep_core::{
AuthorizationCarrier, Command, CredentialScheme, ErrorCode, command_path_from_inspect,
parse_protected_resource_authorization,
};
use aep_service::{AuthenticatedCommandOptions, IdempotentCommandOptions, Service};
use bytes::{Buf, Bytes};
use futures::{FutureExt as _, future::BoxFuture};
use http::{HeaderMap, Method, Request, StatusCode, header};
use http_body::Body;
use http_body_util::{BodyExt as _, LengthLimitError, Limited};
use tower::Service as TowerService;
use url::Url;
use crate::{
HttpResponse, TowerError,
response::{
empty_response, head_response, internal_error, json_response, method_not_allowed,
problem_response, service_response,
},
};
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct CommandPaths {
pub enroll: String,
pub grant: String,
pub inspect: String,
pub revoke: String,
pub status: String,
}
impl CommandPaths {
fn new(service: &Service) -> Result<Self, TowerError> {
let inspect = service.inspect_document();
let origin = Url::parse("https://aep.invalid/")?;
let path = |command| -> Result<String, TowerError> {
Ok(origin
.join(&command_path_from_inspect(&inspect, command)?)?
.path()
.to_owned())
};
Ok(Self {
enroll: path(&Command::Enroll)?,
grant: path(&Command::Grant)?,
inspect: aep_core::WELL_KNOWN_PATH.to_owned(),
revoke: path(&Command::Revoke)?,
status: path(&Command::Status)?,
})
}
}
#[derive(Clone)]
pub struct CommandService {
maximum_request_bytes: usize,
paths: CommandPaths,
service: Arc<Service>,
}
impl CommandService {
pub fn new(service: Arc<Service>, maximum_request_bytes: usize) -> Result<Self, TowerError> {
if maximum_request_bytes == 0 {
return Err(TowerError::InvalidConfiguration(
"request body limit must be positive".to_owned(),
));
}
Ok(Self {
maximum_request_bytes,
paths: CommandPaths::new(&service)?,
service,
})
}
pub const fn paths(&self) -> &CommandPaths {
&self.paths
}
pub async fn dispatch<B>(&self, request: Request<B>) -> Result<HttpResponse, TowerError>
where
B: Body + Send + 'static,
B::Data: Buf + Send,
B::Error: Into<Box<dyn Error + Send + Sync>>,
{
let head = request.method() == Method::HEAD;
let command = match self.route(request.method(), request.uri().path()) {
Route::Unknown => return Ok(empty_response(StatusCode::NOT_FOUND)),
Route::MethodNotAllowed(allow) => {
let response = method_not_allowed(allow);
return Ok(if head {
head_response(response)
} else {
response
});
}
Route::Inspect => {
let response = json_response(
StatusCode::OK,
aep_core::MEDIA_TYPE,
&self.service.inspect_document(),
)?;
return Ok(if head {
head_response(response)
} else {
response
});
}
Route::Status => {
let response = self
.service
.status(AuthenticatedCommandOptions {
client_assertion: client_assertion(request.headers()),
})
.await?;
let response = service_response(response)?;
return Ok(if head {
head_response(response)
} else {
response
});
}
Route::Command(command) => command,
};
if !is_aep_content_type(request.headers()) {
return Ok(problem_response(
ErrorCode::InvalidRequest,
StatusCode::BAD_REQUEST,
));
}
let client_assertion = client_assertion(request.headers());
let idempotency_key = header_value(request.headers(), "idempotency-key");
let body = collect_body(request.into_body(), self.maximum_request_bytes).await?;
let options = IdempotentCommandOptions {
client_assertion,
idempotency_key,
};
let response = match command {
CommandRoute::Enroll => self.service.enroll(&body, options).await,
CommandRoute::Grant => self.service.grant(&body, options).await,
CommandRoute::Revoke => self.service.revoke(&body, options).await,
}?;
service_response(response)
}
fn route(&self, method: &Method, path: &str) -> Route {
for (candidate, expected, route) in [
(self.paths.inspect.as_str(), Method::GET, Route::Inspect),
(
self.paths.enroll.as_str(),
Method::POST,
Route::Command(CommandRoute::Enroll),
),
(self.paths.status.as_str(), Method::GET, Route::Status),
(
self.paths.grant.as_str(),
Method::POST,
Route::Command(CommandRoute::Grant),
),
(
self.paths.revoke.as_str(),
Method::POST,
Route::Command(CommandRoute::Revoke),
),
] {
if path != candidate {
continue;
}
let safe = expected == Method::GET;
return if method == expected || (safe && method == Method::HEAD) {
route
} else if safe {
Route::MethodNotAllowed("GET, HEAD")
} else {
Route::MethodNotAllowed("POST")
};
}
Route::Unknown
}
}
impl<B> TowerService<Request<B>> for CommandService
where
B: Body + Send + 'static,
B::Data: Buf + Send,
B::Error: Into<Box<dyn Error + Send + Sync>>,
{
type Response = HttpResponse;
type Error = Infallible;
type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
fn poll_ready(&mut self, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
Poll::Ready(Ok(()))
}
fn call(&mut self, request: Request<B>) -> Self::Future {
let service = self.clone();
async move {
let response = match service.dispatch(request).await {
Ok(response) => response,
Err(TowerError::RequestTooLarge) => {
problem_response(ErrorCode::InvalidRequest, StatusCode::PAYLOAD_TOO_LARGE)
}
Err(_) => internal_error(),
};
Ok(response)
}
.boxed()
}
}
#[derive(Clone, Copy)]
enum Route {
Command(CommandRoute),
Inspect,
MethodNotAllowed(&'static str),
Status,
Unknown,
}
#[derive(Clone, Copy)]
enum CommandRoute {
Enroll,
Grant,
Revoke,
}
async fn collect_body<B>(body: B, maximum_bytes: usize) -> Result<Bytes, TowerError>
where
B: Body,
B::Data: Buf,
B::Error: Into<Box<dyn Error + Send + Sync>>,
{
Limited::new(body, maximum_bytes)
.collect()
.await
.map(|body| body.to_bytes())
.map_err(|error| {
if error.downcast_ref::<LengthLimitError>().is_some() {
TowerError::RequestTooLarge
} else {
TowerError::RequestBody(error.to_string())
}
})
}
fn client_assertion(headers: &HeaderMap) -> String {
let values = headers
.get_all(header::AUTHORIZATION)
.iter()
.collect::<Vec<_>>();
if values.len() != 1 {
return String::new();
}
let Ok(value) = values[0].to_str() else {
return String::new();
};
let Ok(parsed) = parse_protected_resource_authorization(value, AuthorizationCarrier::Standard)
else {
return String::new();
};
if parsed.scheme == CredentialScheme::Aep {
parsed.credentials
} else {
String::new()
}
}
fn header_value(headers: &HeaderMap, name: &str) -> String {
let values = headers.get_all(name).iter().collect::<Vec<_>>();
if values.len() == 1 {
values[0].to_str().unwrap_or_default().to_owned()
} else {
String::new()
}
}
fn is_aep_content_type(headers: &HeaderMap) -> bool {
let values = headers
.get_all(header::CONTENT_TYPE)
.iter()
.collect::<Vec<_>>();
values.len() == 1
&& values[0].to_str().is_ok_and(|value| {
value.split(';').next().is_some_and(|media_type| {
media_type.trim().eq_ignore_ascii_case(aep_core::MEDIA_TYPE)
})
})
}