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;
19use url::Url;
20
21use crate::{
22    HttpResponse, TowerError,
23    response::{
24        empty_response, head_response, internal_error, json_response, method_not_allowed,
25        problem_response, service_response,
26    },
27};
28
29/// Percent-encoded HTTP request paths for mounting the command endpoints.
30#[derive(Clone, Debug, Eq, PartialEq)]
31pub struct CommandPaths {
32    pub enroll: String,
33    pub grant: String,
34    pub inspect: String,
35    pub revoke: String,
36    pub status: String,
37}
38
39impl CommandPaths {
40    fn new(service: &Service) -> Result<Self, TowerError> {
41        let inspect = service.inspect_document();
42        // Match the Agent's URL encoding before comparing HTTP request paths.
43        let origin = Url::parse("https://aep.invalid/")?;
44        let path = |command| -> Result<String, TowerError> {
45            Ok(origin
46                .join(&command_path_from_inspect(&inspect, command)?)?
47                .path()
48                .to_owned())
49        };
50        Ok(Self {
51            enroll: path(&Command::Enroll)?,
52            grant: path(&Command::Grant)?,
53            inspect: aep_core::WELL_KNOWN_PATH.to_owned(),
54            revoke: path(&Command::Revoke)?,
55            status: path(&Command::Status)?,
56        })
57    }
58}
59
60#[derive(Clone)]
61pub struct CommandService {
62    maximum_request_bytes: usize,
63    paths: CommandPaths,
64    service: Arc<Service>,
65}
66
67impl CommandService {
68    pub fn new(service: Arc<Service>, maximum_request_bytes: usize) -> Result<Self, TowerError> {
69        if maximum_request_bytes == 0 {
70            return Err(TowerError::InvalidConfiguration(
71                "request body limit must be positive".to_owned(),
72            ));
73        }
74        Ok(Self {
75            maximum_request_bytes,
76            paths: CommandPaths::new(&service)?,
77            service,
78        })
79    }
80
81    pub const fn paths(&self) -> &CommandPaths {
82        &self.paths
83    }
84
85    pub async fn dispatch<B>(&self, request: Request<B>) -> Result<HttpResponse, TowerError>
86    where
87        B: Body + Send + 'static,
88        B::Data: Buf + Send,
89        B::Error: Into<Box<dyn Error + Send + Sync>>,
90    {
91        // RFC 9110 Section 9.3.2 requires HEAD wherever GET is supported, and answers it with the
92        // GET response minus the content.
93        let head = request.method() == Method::HEAD;
94        let command = match self.route(request.method(), request.uri().path()) {
95            Route::Unknown => return Ok(empty_response(StatusCode::NOT_FOUND)),
96            Route::MethodNotAllowed(allow) => {
97                let response = method_not_allowed(allow);
98                return Ok(if head {
99                    head_response(response)
100                } else {
101                    response
102                });
103            }
104            Route::Inspect => {
105                let response = json_response(
106                    StatusCode::OK,
107                    aep_core::MEDIA_TYPE,
108                    &self.service.inspect_document(),
109                )?;
110                return Ok(if head {
111                    head_response(response)
112                } else {
113                    response
114                });
115            }
116            Route::Status => {
117                let response = self
118                    .service
119                    .status(AuthenticatedCommandOptions {
120                        client_assertion: client_assertion(request.headers()),
121                    })
122                    .await?;
123                let response = service_response(response)?;
124                return Ok(if head {
125                    head_response(response)
126                } else {
127                    response
128                });
129            }
130            Route::Command(command) => command,
131        };
132        if !is_aep_content_type(request.headers()) {
133            return Ok(problem_response(
134                ErrorCode::InvalidRequest,
135                StatusCode::BAD_REQUEST,
136            ));
137        }
138        let client_assertion = client_assertion(request.headers());
139        let idempotency_key = header_value(request.headers(), "idempotency-key");
140        let body = collect_body(request.into_body(), self.maximum_request_bytes).await?;
141        let options = IdempotentCommandOptions {
142            client_assertion,
143            idempotency_key,
144        };
145        let response = match command {
146            CommandRoute::Enroll => self.service.enroll(&body, options).await,
147            CommandRoute::Grant => self.service.grant(&body, options).await,
148            CommandRoute::Revoke => self.service.revoke(&body, options).await,
149        }?;
150        service_response(response)
151    }
152
153    fn route(&self, method: &Method, path: &str) -> Route {
154        for (candidate, expected, route) in [
155            (self.paths.inspect.as_str(), Method::GET, Route::Inspect),
156            (
157                self.paths.enroll.as_str(),
158                Method::POST,
159                Route::Command(CommandRoute::Enroll),
160            ),
161            (self.paths.status.as_str(), Method::GET, Route::Status),
162            (
163                self.paths.grant.as_str(),
164                Method::POST,
165                Route::Command(CommandRoute::Grant),
166            ),
167            (
168                self.paths.revoke.as_str(),
169                Method::POST,
170                Route::Command(CommandRoute::Revoke),
171            ),
172        ] {
173            if path != candidate {
174                continue;
175            }
176            let safe = expected == Method::GET;
177            return if method == expected || (safe && method == Method::HEAD) {
178                route
179            } else if safe {
180                Route::MethodNotAllowed("GET, HEAD")
181            } else {
182                Route::MethodNotAllowed("POST")
183            };
184        }
185        Route::Unknown
186    }
187}
188
189impl<B> TowerService<Request<B>> for CommandService
190where
191    B: Body + Send + 'static,
192    B::Data: Buf + Send,
193    B::Error: Into<Box<dyn Error + Send + Sync>>,
194{
195    type Response = HttpResponse;
196    type Error = Infallible;
197    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
198
199    fn poll_ready(&mut self, _context: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
200        Poll::Ready(Ok(()))
201    }
202
203    fn call(&mut self, request: Request<B>) -> Self::Future {
204        let service = self.clone();
205        async move {
206            let response = match service.dispatch(request).await {
207                Ok(response) => response,
208                Err(TowerError::RequestTooLarge) => {
209                    problem_response(ErrorCode::InvalidRequest, StatusCode::PAYLOAD_TOO_LARGE)
210                }
211                Err(_) => internal_error(),
212            };
213            Ok(response)
214        }
215        .boxed()
216    }
217}
218
219/// A resolved AEP endpoint.
220///
221/// The three body-carrying commands share one variant so that the arm selecting the Service call
222/// is total: an endpoint cannot be added to the router without the compiler demanding its call.
223#[derive(Clone, Copy)]
224enum Route {
225    Command(CommandRoute),
226    Inspect,
227    MethodNotAllowed(&'static str),
228    Status,
229    Unknown,
230}
231
232#[derive(Clone, Copy)]
233enum CommandRoute {
234    Enroll,
235    Grant,
236    Revoke,
237}
238
239async fn collect_body<B>(body: B, maximum_bytes: usize) -> Result<Bytes, TowerError>
240where
241    B: Body,
242    B::Data: Buf,
243    B::Error: Into<Box<dyn Error + Send + Sync>>,
244{
245    Limited::new(body, maximum_bytes)
246        .collect()
247        .await
248        .map(|body| body.to_bytes())
249        .map_err(|error| {
250            if error.downcast_ref::<LengthLimitError>().is_some() {
251                TowerError::RequestTooLarge
252            } else {
253                TowerError::RequestBody(error.to_string())
254            }
255        })
256}
257
258fn client_assertion(headers: &HeaderMap) -> String {
259    let values = headers
260        .get_all(header::AUTHORIZATION)
261        .iter()
262        .collect::<Vec<_>>();
263    if values.len() != 1 {
264        return String::new();
265    }
266    let Ok(value) = values[0].to_str() else {
267        return String::new();
268    };
269    let Ok(parsed) = parse_protected_resource_authorization(value, AuthorizationCarrier::Standard)
270    else {
271        return String::new();
272    };
273    if parsed.scheme == CredentialScheme::Aep {
274        parsed.credentials
275    } else {
276        String::new()
277    }
278}
279
280fn header_value(headers: &HeaderMap, name: &str) -> String {
281    let values = headers.get_all(name).iter().collect::<Vec<_>>();
282    if values.len() == 1 {
283        values[0].to_str().unwrap_or_default().to_owned()
284    } else {
285        String::new()
286    }
287}
288
289fn is_aep_content_type(headers: &HeaderMap) -> bool {
290    let values = headers
291        .get_all(header::CONTENT_TYPE)
292        .iter()
293        .collect::<Vec<_>>();
294    values.len() == 1
295        && values[0].to_str().is_ok_and(|value| {
296            value.split(';').next().is_some_and(|media_type| {
297                media_type.trim().eq_ignore_ascii_case(aep_core::MEDIA_TYPE)
298            })
299        })
300}
aep_tower/command.rs

aep_tower/
command.rs
