activitypub-federation-rust/src/axum/inbox.rs

90 lines
2.7 KiB
Rust
Raw Normal View History

2023-03-05 17:17:34 -08:00
//! Handles incoming activities, verifying HTTP signatures and other checks
//!
#![doc = include_str!("../../docs/08_receiving_activities.md")]
use crate::{
config::Data,
2023-03-01 15:19:10 -08:00
error::Error,
http_signatures::verify_signature,
parse_received_activity,
2023-03-16 13:41:29 -07:00
traits::{ActivityHandler, Actor, Object},
};
2023-03-05 17:17:34 -08:00
use axum::{
body::Body,
extract::{FromRequest, FromRequestParts, OriginalUri},
2023-03-05 17:17:34 -08:00
http::{Request, StatusCode},
response::{IntoResponse, Response},
};
use http::{HeaderMap, Method};
use serde::de::DeserializeOwned;
use tracing::debug;
2023-03-02 06:18:06 -08:00
/// Handles incoming activities, verifying HTTP signatures and other checks
pub async fn receive_activity<Activity, ActorT, Datatype>(
activity_data: ActivityData,
data: &Data<Datatype>,
) -> Result<(), <Activity as ActivityHandler>::Error>
where
Activity: ActivityHandler<DataType = Datatype> + DeserializeOwned + Send + 'static,
2023-03-16 13:41:29 -07:00
ActorT: Object<DataType = Datatype> + Actor + Send + 'static,
for<'de2> <ActorT as Object>::Kind: serde::Deserialize<'de2>,
<Activity as ActivityHandler>::Error: From<Error> + From<<ActorT as Object>::Error>,
<ActorT as Object>::Error: From<Error>,
Datatype: Clone,
{
let (activity, actor) =
parse_received_activity::<Activity, ActorT, _>(&activity_data.body, data).await?;
verify_signature(
&activity_data.headers,
&activity_data.method,
&activity_data.uri,
actor.public_key_pem(),
)?;
debug!("Receiving activity {}", activity.id().to_string());
activity.verify(data).await?;
activity.receive(data).await?;
Ok(())
}
2023-03-05 17:17:34 -08:00
/// Contains all data that is necessary to receive an activity from an HTTP request
#[derive(Debug)]
pub struct ActivityData {
headers: HeaderMap,
method: Method,
uri: OriginalUri,
2023-03-05 17:17:34 -08:00
body: Vec<u8>,
}
impl<S> FromRequest<S> for ActivityData
2023-03-05 17:17:34 -08:00
where
S: Send + Sync,
{
type Rejection = Response;
async fn from_request(req: Request<Body>, state: &S) -> Result<Self, Self::Rejection> {
let (mut parts, body) = req.into_parts();
// take the full URI to handle nested routers
// OriginalUri::from_request_parts has an Infallible error type
let uri = OriginalUri::from_request_parts(&mut parts, state)
.await
.expect("infallible");
2023-03-05 17:17:34 -08:00
// this wont work if the body is an long running stream
let bytes = axum::body::to_bytes(body, usize::MAX)
2023-03-05 17:17:34 -08:00
.await
.map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response())?;
Ok(Self {
headers: parts.headers,
method: parts.method,
uri,
2023-03-05 17:17:34 -08:00
body: bytes.to_vec(),
})
}
}
// TODO: copy tests from actix-web inbox and implement for axum as well