aep_tower/
command.rs

1use std::{
2    convert::Infallible,
3    error::Error,
4    sync::Arc,
5    task::{Context, Poll},
6};
7
8use aep_core::{
9    AuthorizationCarrier, Command, CredentialScheme, ErrorCode, command_path_from_inspect,
10    parse_protected_resource_authorization,
11};
12use aep_service::{AuthenticatedCommandOptions, IdempotentCommandOptions, Service};
13use bytes::{Buf, Bytes};
14use futures::{FutureExt as _, future::BoxFuture};
15use http::{HeaderMap, Method, Request, StatusCode, header};
16use http_body::Body;
17use http_body_util::{BodyExt as _, LengthLimitError, Limited};
18use tower::Service as TowerService;
19
20use crate::{
21    HttpResponse, TowerError,
22    response::{
23        empty_response, internal_error, json_response, method_not_allowed, problem_response,
24        service_response,
25    },
26};
27
28#[derive(Clone, Debug, Eq, PartialEq)]
29pub struct CommandPaths {
30    pub enroll: String,
31    pub grant: String,
32    pub inspect: String,
33    pub revoke: String,
34    pub status: String,
35}
36
37impl CommandPaths {
38    fn new(service: &Service) -> Result<Self, TowerError> {
39        let inspect = service.inspect_document();
40        Ok(Self {
41            enroll: command_path_from_inspect(&inspect, &Command::Enroll)?,
42            grant: command_path_from_inspect(&inspect, &Command::Grant)?,
43            inspect: aep_core::WELL_KNOWN_PATH.to_owned(),
44            revoke: command_path_from_inspect(&inspect, &Command::Revoke)?,
45            status: command_path_from_inspect(&inspect, &Command::Status)?,
46        })
47    }
48}
49
50#[derive(Clone)]
51pub struct CommandService {
52    maximum_request_bytes: usize,
53    paths: CommandPaths,
54    service: Arc<Service>,
55}
56
57impl CommandService {
58    pub fn new(service: Arc<Service>, maximum_request_bytes: usize) -> Result<Self, TowerError> {
59        if maximum_request_bytes == 0 {
60            return Err(TowerError::InvalidConfiguration(
61                "request body limit must be positive".to_owned(),
62            ));
63        }
64        Ok(Self {
65            maximum_request_bytes,
66            paths: CommandPaths::new(&service)?,
67            service,
68        })
69    }
70
71    pub const fn paths(&self) -> &CommandPaths {
72        &self.paths
73    }
74
75    pub async fn dispatch<B>(&self, request: Request<B>) -> Result<HttpResponse, TowerError>
76    where
77        B: Body + Send + 'static,
78        B::Data: Buf + Send,
79        B::Error: Into<Box<dyn Error + Send + Sync>>,
80    {
81        let route = self.route(request.method(), request.uri().path());
82        match route {
83            Route::Unknown => return Ok(empty_response(StatusCode::NOT_FOUND)),
84            Route::MethodNotAllowed(method) => return Ok(method_not_allowed(method)),
85            Route::Inspect => {
86                return json_response(
87                    StatusCode::OK,
88                    aep_core::MEDIA_TYPE,
89                    &self.service.inspect_document(),
90                );
91            }
92            Route::Status => {
93                let response = self
94                    .service
95                    .status(AuthenticatedCommandOptions {
96                        client_assertion: client_assertion(request.headers()),
97                    })
98                    .await?;
99                return service_response(response);
100            }
101            Route::Enroll | Route::Grant | Route::Revoke => {}
102        }
103        if !is_aep_content_type(request.headers()) {
104            return Ok(problem_response(
105                ErrorCode::InvalidRequest,
106                StatusCode::BAD_REQUEST,
107            ));
108        }
109        let client_assertion = client_assertion(request.headers());
110        let idempotency_key = header_value(request.headers(), "idempotency-key");
111        let body = collect_body(request.into_body(), self.maximum_request_bytes).await?;
112        let options = IdempotentCommandOptions {
113            client_assertion,
114            idempotency_key,
115        };
116        let response = match route {
117            Route::Enroll => self.service.enroll(&body, options).await,
118            Route::Grant => self.service.grant(&body, options).await,
119            Route::Revoke => self.service.revoke(&body, options).await,
120            Route::Inspect | Route::Status | Route::Unknown | Route::MethodNotAllowed(_) => {
121                unreachable!()
122            }
123        }?;
124        service_response(response)
125    }
126
127    fn route(&self, method: &Method, path: &str) -> Route {
128        for (candidate, expected, route) in [
129            (self.paths.inspect.as_str(), Method::GET, Route::Inspect),
130            (self.paths.enroll.as_str(), Method::POST, Route::Enroll),
131            (self.paths.status.as_str(), Method::GET, Route::Status),
132            (self.paths.grant.as_str(), Method::POST, Route::Grant),
133            (self.paths.revoke.as_str(), Method::POST, Route::Revoke),
134        ] {
135            if path == candidate {
136                return if method == expected {
137                    route
138                } else {
139                    Route::MethodNotAllowed(if expected == Method::GET {
140                        "GET"
141                    } else {
142                        "POST"
143                    })
144                };
145            }
146        }
147        Route::Unknown
148    }
149}
150
151impl<B> TowerService<Request<B>> for CommandService
152where
153    B: Body + Send + 'static,
154    B::Data: Buf + Send,
155    B::Error: Into<Box<dyn Error + Send + Sync>>,
156{
157    type Response = HttpResponse;
158    type Error = Infallible;
159    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
160
161    fn poll_ready(&mut self, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
162        Poll::Ready(Ok(()))
163    }
164
165    fn call(&mut self, request: Request<B>) -> Self::Future {
166        let service = self.clone();
167        async move {
168            let response = match service.dispatch(request).await {
169                Ok(response) => response,
170                Err(TowerError::RequestTooLarge) => {
171                    problem_response(ErrorCode::InvalidRequest, StatusCode::PAYLOAD_TOO_LARGE)
172                }
173                Err(_) => internal_error(),
174            };
175            Ok(response)
176        }
177        .boxed()
178    }
179}
180
181#[derive(Clone, Copy)]
182enum Route {
183    Enroll,
184    Grant,
185    Inspect,
186    MethodNotAllowed(&'static str),
187    Revoke,
188    Status,
189    Unknown,
190}
191
192async fn collect_body<B>(body: B, maximum_bytes: usize) -> Result<Bytes, TowerError>
193where
194    B: Body,
195    B::Data: Buf,
196    B::Error: Into<Box<dyn Error + Send + Sync>>,
197{
198    Limited::new(body, maximum_bytes)
199        .collect()
200        .await
201        .map(|body| body.to_bytes())
202        .map_err(|error| {
203            if error.downcast_ref::<LengthLimitError>().is_some() {
204                TowerError::RequestTooLarge
205            } else {
206                TowerError::RequestBody(error.to_string())
207            }
208        })
209}
210
211fn client_assertion(headers: &HeaderMap) -> String {
212    let values = headers
213        .get_all(header::AUTHORIZATION)
214        .iter()
215        .collect::<Vec<_>>();
216    if values.len() != 1 {
217        return String::new();
218    }
219    let Ok(value) = values[0].to_str() else {
220        return String::new();
221    };
222    let Ok(parsed) = parse_protected_resource_authorization(value, AuthorizationCarrier::Standard)
223    else {
224        return String::new();
225    };
226    if parsed.scheme == CredentialScheme::Aep {
227        parsed.credentials
228    } else {
229        String::new()
230    }
231}
232
233fn header_value(headers: &HeaderMap, name: &str) -> String {
234    let values = headers.get_all(name).iter().collect::<Vec<_>>();
235    if values.len() == 1 {
236        values[0].to_str().unwrap_or_default().to_owned()
237    } else {
238        String::new()
239    }
240}
241
242fn is_aep_content_type(headers: &HeaderMap) -> bool {
243    let values = headers
244        .get_all(header::CONTENT_TYPE)
245        .iter()
246        .collect::<Vec<_>>();
247    values.len() == 1
248        && values[0].to_str().is_ok_and(|value| {
249            value.split(';').next().is_some_and(|media_type| {
250                media_type.trim().eq_ignore_ascii_case(aep_core::MEDIA_TYPE)
251            })
252        })
253}
aep_tower/command.rs

aep_tower/
command.rs
