Repository: cloudflare/daphne
Pull request #685 (2024-09-24 15:49:49)

Refactor DapRequest in order to extract parsing code out of protocol logic
- use DapError in DapExtractor
- Move DapRequest and it's extractors to their own modules
- Split the request into payload and metadata.
- Parameterize DapRequest by its payload type



Patch:
diff --git a/crates/dapf/src/main.rs b/crates/dapf/src/main.rs
index b41bff5..c82ab48 100644
--- a/crates/dapf/src/main.rs
+++ b/crates/dapf/src/main.rs
@@ -524,7 +524,7 @@ async fn handle_leader_actions(
             headers.insert(
                 reqwest::header::CONTENT_TYPE,
                 reqwest::header::HeaderValue::from_str(
-                    DapMediaType::CollectReq
+                    DapMediaType::CollectionReq
                         .as_str_for_version(version)
                         .ok_or_else(|| anyhow!("invalid content-type for dap version"))?,
                 )
diff --git a/crates/daphne-server/src/lib.rs b/crates/daphne-server/src/lib.rs
index cbd4210..75ae51a 100644
--- a/crates/daphne-server/src/lib.rs
+++ b/crates/daphne-server/src/lib.rs
@@ -164,7 +164,7 @@ impl router::DaphneService for App {
         }
     }
 
-    async fn is_using_taskprov(&self, req: &daphne::DapRequest) -> Result<bool, DapError> {
+    async fn is_using_taskprov(&self, req: &daphne::DapRequestMeta) -> Result<bool, DapError> {
         if req.taskprov.is_some() {
             Ok(true)
         } else if self
diff --git a/crates/daphne-server/src/roles/aggregator.rs b/crates/daphne-server/src/roles/aggregator.rs
index 28a3a04..78a53e0 100644
--- a/crates/daphne-server/src/roles/aggregator.rs
+++ b/crates/daphne-server/src/roles/aggregator.rs
@@ -13,7 +13,8 @@ use daphne::{
     metrics::DaphneMetrics,
     roles::{aggregator::MergeAggShareError, DapAggregator, DapReportInitializer},
     taskprov, DapAggregateShare, DapAggregateSpan, DapAggregationParam, DapError, DapGlobalConfig,
-    DapRequest, DapTaskConfig, DapVersion, EarlyReportStateConsumed, EarlyReportStateInitialized,
+    DapRequestMeta, DapTaskConfig, DapVersion, EarlyReportStateConsumed,
+    EarlyReportStateInitialized,
 };
 use daphne_service_utils::durable_requests::bindings::{
     self, AggregateStoreMergeOptions, AggregateStoreMergeReq, AggregateStoreMergeResp,
@@ -226,7 +227,7 @@ impl DapAggregator for crate::App {
 
     async fn taskprov_put(
         &self,
-        req: &DapRequest,
+        req: &DapRequestMeta,
         task_config: DapTaskConfig,
     ) -> Result<(), DapError> {
         let task_id = &req.task_id;
diff --git a/crates/daphne-server/src/roles/leader.rs b/crates/daphne-server/src/roles/leader.rs
index a6bc0d7..eff3d49 100644
--- a/crates/daphne-server/src/roles/leader.rs
+++ b/crates/daphne-server/src/roles/leader.rs
@@ -12,10 +12,11 @@ use daphne::{
     fatal_error,
     messages::{BatchId, BatchSelector, Collection, CollectionJobId, Report, TaskId},
     roles::{leader::WorkItem, DapAggregator, DapLeader},
-    DapAggregationParam, DapCollectionJob, DapError, DapRequest, DapResponse,
+    DapAggregationParam, DapCollectionJob, DapError, DapRequest, DapResponse, DapVersion,
 };
 use daphne_service_utils::http_headers;
 use http::StatusCode;
+use prio::codec::ParameterizedEncode;
 use tracing::{error, info};
 use url::Url;
 
@@ -99,22 +100,31 @@ impl DapLeader for crate::App {
         self.test_leader_state.lock().await.enqueue_work(items)
     }
 
-    async fn send_http_post(&self, req: DapRequest, url: Url) -> Result<DapResponse, DapError> {
+    async fn send_http_post<M>(&self, req: DapRequest<M>, url: Url) -> Result<DapResponse, DapError>
+    where
+        M: Send + ParameterizedEncode<DapVersion>,
+    {
         self.send_http(req, Method::POST, url).await
     }
 
-    async fn send_http_put(&self, req: DapRequest, url: Url) -> Result<DapResponse, DapError> {
+    async fn send_http_put<M>(&self, req: DapRequest<M>, url: Url) -> Result<DapResponse, DapError>
+    where
+        M: Send + ParameterizedEncode<DapVersion>,
+    {
         self.send_http(req, Method::PUT, url).await
     }
 }
 
 impl crate::App {
-    async fn send_http(
+    async fn send_http<M>(
         &self,
-        req: DapRequest,
+        req: DapRequest<M>,
         method: Method,
         url: Url,
-    ) -> Result<DapResponse, DapError> {
+    ) -> Result<DapResponse, DapError>
+    where
+        M: Send + ParameterizedEncode<DapVersion>,
+    {
         use reqwest::header::{self, HeaderMap, HeaderName, HeaderValue};
         let content_type = req
             .media_type
@@ -181,10 +191,15 @@ impl crate::App {
             );
         }
 
+        let DapRequest { meta, payload } = req;
         let req_builder = self
             .http
             .request(method, url.clone())
-            .body(req.payload)
+            .body(
+                payload
+                    .get_encoded_with_param(&meta.version)
+                    .map_err(|e| DapAbort::from_codec_error(e, meta.task_id))?,
+            )
             .headers(headers);
 
         let start = Instant::now();
@@ -202,7 +217,7 @@ impl crate::App {
                 .get_all(reqwest::header::CONTENT_TYPE)
                 .into_iter()
                 .filter_map(|h| h.to_str().ok())
-                .find_map(|h| DapMediaType::from_str_for_version(req.version, h))
+                .find_map(|h| DapMediaType::from_str_for_version(meta.version, h))
                 .ok_or_else(|| fatal_error!(err = "peer response is missing media type"))?;
 
             let payload = reqwest_resp
@@ -212,7 +227,7 @@ impl crate::App {
                 .to_vec();
 
             Ok(DapResponse {
-                version: req.version,
+                version: meta.version,
                 payload,
                 media_type,
             })
@@ -236,7 +251,7 @@ impl crate::App {
                 StatusCode::UNAUTHORIZED => {
                     return Err(DapAbort::UnauthorizedRequest {
                         detail: format!("helper at {url} didn't authorize our request"),
-                        task_id: req.task_id,
+                        task_id: meta.task_id,
                     }
                     .into())
                 }
diff --git a/crates/daphne-server/src/router/extractor.rs b/crates/daphne-server/src/router/extractor.rs
new file mode 100644
index 0000000..5439a87
--- /dev/null
+++ b/crates/daphne-server/src/router/extractor.rs
@@ -0,0 +1,599 @@
+// Copyright (c) 2024 Cloudflare, Inc. All rights reserved.
+// SPDX-License-Identifier: BSD-3-Clause
+
+use std::io::Cursor;
+
+use axum::{
+    async_trait,
+    body::{Bytes, HttpBody},
+    extract::{FromRequest, FromRequestParts, Path},
+};
+use daphne::{
+    constants::DapMediaType,
+    error::DapAbort,
+    fatal_error,
+    messages::{
+        AggregateShareReq, AggregationJobId, AggregationJobInitReq, CollectionJobId, CollectionReq,
+        Report, TaskId,
+    },
+    DapError, DapRequest, DapRequestMeta, DapResource, DapVersion,
+};
+use daphne_service_utils::{bearer_token::BearerToken, http_headers, metrics};
+use http::{header::CONTENT_TYPE, HeaderMap, Request};
+use prio::codec::ParameterizedDecode;
+use serde::Deserialize;
+
+use super::{AxumDapResponse, DaphneService};
+
+pub trait DecodeFromDapHttpBody: Sized {
+    fn decode_from_http_body(bytes: Bytes, meta: &DapRequestMeta) -> Result<Self, DapAbort>;
+}
+
+macro_rules! impl_decode_from_dap_http_body {
+    ($($type:ident),*$(,)?) => {
+        $(
+            impl DecodeFromDapHttpBody for $type {
+                fn decode_from_http_body(
+                    bytes: Bytes,
+                    meta: &DapRequestMeta,
+                ) -> Result<Self, DapAbort> {
+                    let mut cursor = Cursor::new(bytes.as_ref());
+                    meta.get_checked_media_type(DapMediaType::$type)?;
+                    $type::decode_with_param(&meta.version, &mut cursor)
+                        .map_err(|e| DapAbort::from_codec_error(e, meta.task_id))
+                }
+            }
+        )*
+    };
+}
+
+impl_decode_from_dap_http_body!(
+    AggregationJobInitReq,
+    AggregateShareReq,
+    Report,
+    CollectionReq,
+);
+
+impl DecodeFromDapHttpBody for () {
+    fn decode_from_http_body(_bytes: Bytes, _meta: &DapRequestMeta) -> Result<Self, DapAbort> {
+        Ok(())
+    }
+}
+
+/// An axum extractor capable of parsing a [`DapRequest`].
+#[derive(Debug)]
+pub(super) struct UnauthenticatedDapRequestExtractor<P>(pub DapRequest<P>);
+
+#[async_trait]
+impl<S, B, P> FromRequest<S, B, P> for UnauthenticatedDapRequestExtractor<P>
+where
+    P: DecodeFromDapHttpBody,
+    S: DaphneService + Send + Sync,
+    B: HttpBody + Send + 'static,
+    <B as HttpBody>::Data: Send,
+{
+    type Rejection = AxumDapResponse;
+
+    async fn from_request(req: Request<B>, state: &S) -> Result<Self, Self::Rejection> {
+        #[derive(Debug, Deserialize)]
+        #[serde(deny_unknown_fields)]
+        struct PathParams {
+            version: DapVersion,
+            #[serde(default, with = "daphne::messages::base64url")]
+            task_id: TaskId,
+            #[serde(default, with = "daphne::messages::base64url_option")]
+            agg_job_id: Option<AggregationJobId>,
+            #[serde(default, with = "daphne::messages::base64url_option")]
+            collect_job_id: Option<CollectionJobId>,
+        }
+
+        let (mut parts, body) = req.into_parts();
+        let Path(PathParams {
+            version,
+            task_id,
+            agg_job_id,
+            collect_job_id,
+        }) = Path::from_request_parts(&mut parts, state)
+            .await
+            .map_err(|_| {
+                AxumDapResponse::new_error(
+                    DapAbort::BadRequest("invalid path".into()),
+                    state.server_metrics(),
+                )
+            })?;
+
+        let media_type = if let Some(content_type) = parts.headers.get(CONTENT_TYPE) {
+            let content_type = content_type.to_str().map_err(|_| {
+                let msg = "header value contains non ascii or invisible characters".into();
+                AxumDapResponse::new_error(DapAbort::BadRequest(msg), state.server_metrics())
+            })?;
+            let media_type =
+                DapMediaType::from_str_for_version(version, content_type).ok_or_else(|| {
+                    AxumDapResponse::new_error(
+                        DapAbort::BadRequest("invalid media type".into()),
+                        state.server_metrics(),
+                    )
+                })?;
+            Some(media_type)
+        } else {
+            None
+        };
+
+        // TODO(mendess): this allocates needlessly, if prio supported some kind of
+        // AsyncParameterizedDecode we could avoid this allocation
+        let payload = hyper::body::to_bytes(body).await;
+
+        let Ok(payload) = payload else {
+            return Err(AxumDapResponse::new_error(
+                fatal_error!(err = "failed to get payload"),
+                state.server_metrics(),
+            ));
+        };
+
+        let (task_id, resource) = {
+            let resource = match media_type {
+                Some(DapMediaType::AggregationJobInitReq) => DapResource::AggregationJob(
+                    agg_job_id.expect("route definition should guarantee that agg_job_id exists"),
+                ),
+                Some(DapMediaType::CollectionReq) => DapResource::CollectionJob(
+                    collect_job_id
+                        .expect("route definition should guarantee that collect_job_id exists"),
+                ),
+                _ => DapResource::Undefined,
+            };
+
+            (task_id, resource)
+        };
+
+        let meta = DapRequestMeta {
+            version,
+            task_id,
+            taskprov: extract_header_as_string(&parts.headers, http_headers::DAP_TASKPROV),
+            media_type,
+            resource,
+        };
+
+        let payload = P::decode_from_http_body(payload, &meta)
+            .map_err(|e| AxumDapResponse::new_error(e, state.server_metrics()))?;
+
+        let request = DapRequest { meta, payload };
+
+        Ok(UnauthenticatedDapRequestExtractor(request))
+    }
+}
+
+/// An axum extractor capable of parsing a [`DapRequest`].
+///
+/// This extractor asserts that the request is authenticated.
+#[derive(Debug)]
+pub(super) struct DapRequestExtractor<P>(pub DapRequest<P>);
+
+#[async_trait]
+impl<S, B, P> FromRequest<S, B, P> for DapRequestExtractor<P>
+where
+    P: DecodeFromDapHttpBody + Send + Sync,
+    S: DaphneService + Send + Sync,
+    B: HttpBody + Send + 'static,
+    <B as HttpBody>::Data: Send,
+{
+    type Rejection = AxumDapResponse;
+
+    async fn from_request(req: Request<B>, state: &S) -> Result<Self, Self::Rejection> {
+        let bearer_token = extract_header_as_string(req.headers(), http_headers::DAP_AUTH_TOKEN)
+            .map(BearerToken::from);
+        let cf_tls_client_auth = extract_header_as_string(req.headers(), "X-Client-Cert-Verified");
+
+        let request = UnauthenticatedDapRequestExtractor::from_request(req, state)
+            .await?
+            .0;
+
+        let error_to_response = |error| AxumDapResponse::new_error(error, state.server_metrics());
+        let auth_error = |detail| {
+            error_to_response(DapError::from(DapAbort::UnauthorizedRequest {
+                detail,
+                task_id: request.task_id,
+            }))
+        };
+
+        let is_taskprov = state
+            .is_using_taskprov(&request)
+            .await
+            .map_err(
+                |e| fatal_error!(err = ?e, "failed to determine if request was using taskprov"),
+            )
+            .map_err(error_to_response)?;
+
+        let bearer_authed = if let Some((token, sender)) =
+            bearer_token.zip(request.media_type.map(|m| m.sender()))
+        {
+            state
+                .server_metrics()
+                .auth_method_inc(metrics::AuthMethod::BearerToken);
+            state
+                .check_bearer_token(&token, sender, request.task_id, is_taskprov)
+                .await
+                .map_err(|reason| reason.either(auth_error, error_to_response))?;
+            true
+        } else {
+            false
+        };
+        let mtls_authed = if let Some(verification_result) = cf_tls_client_auth {
+            state
+                .server_metrics()
+                .auth_method_inc(metrics::AuthMethod::TlsClientAuth);
+            // we always check if mtls succedded even if ...
+            if verification_result != "SUCCESS" {
+                return Err(auth_error(format!(
+                    "Invalid TLS certificate ({verification_result})"
+                )));
+            }
+            // ... we only allow mtls auth for taskprov tasks
+            is_taskprov
+        } else {
+            false
+        };
+
+        if bearer_authed || mtls_authed {
+            Ok(Self(request))
+        } else {
+            Err(auth_error(
+                "No suitable authorization method was found".into(),
+            ))
+        }
+    }
+}
+
+fn extract_header_as_string(headers: &HeaderMap, header: &'static str) -> Option<String> {
+    headers.get(header)?.to_str().ok().map(ToString::to_string)
+}
+
+#[cfg(test)]
+mod test {
+    use std::{
+        sync::{Arc, OnceLock},
+        time::Duration,
+    };
+
+    use axum::{
+        body::{Body, HttpBody},
+        extract::State,
+        http::{header::CONTENT_TYPE, Request, StatusCode},
+        response::IntoResponse,
+        routing::get,
+        Router,
+    };
+    use daphne::{
+        async_test_versions,
+        constants::DapMediaType,
+        messages::{AggregationJobId, Base64Encode, CollectionJobId, TaskId},
+        DapError, DapRequestMeta, DapResource, DapSender, DapVersion,
+    };
+    use daphne_service_utils::{
+        bearer_token::BearerToken, http_headers, metrics::DaphnePromServiceMetrics,
+    };
+    use either::Either::{self, Left};
+    use futures::{future::BoxFuture, FutureExt};
+    use rand::{thread_rng, Rng};
+    use tokio::{
+        sync::mpsc::{self, Sender},
+        time::timeout,
+    };
+    use tower::ServiceExt;
+
+    const BEARER_TOKEN: &str = "test-token";
+
+    // In these tests we don't care about the body, so we can type alias it away.
+    type DapRequest = daphne::DapRequest<()>;
+    type UnauthenticatedDapRequestExtractor = super::UnauthenticatedDapRequestExtractor<()>;
+    type DapRequestExtractor = super::DapRequestExtractor<()>;
+
+    /// Return a function that will parse a request using the [`DapRequestExtractor`] or
+    /// [`UnauthenticatedDapRequestExtractor`] and return the parsed request.
+    ///
+    /// The possible request URIs that are supported by this parser are:
+    ///  - `/:version/:task_id/auth` uses the [`DapRequestExtractor`]
+    ///  - `/:version/:task_id/parse-mandatory-fields` uses the [`UnauthenticatedDapRequestExtractor`]
+    ///  - `/:version/:agg_job_id/parse-agg-job-id` uses the [`UnauthenticatedDapRequestExtractor`]
+    ///  - `/:version/:collect_job_id/parse-collect-job-id` uses the [`UnauthenticatedDapRequestExtractor`]
+    fn test_router<B>(
+    ) -> impl FnOnce(Request<B>) -> BoxFuture<'static, Result<DapRequest, StatusCode>>
+    where
+        B: Send + Sync + 'static + HttpBody,
+        B::Data: Send,
+        B::Error: Send + Sync + std::error::Error,
+    {
+        type Channel = Sender<DapRequest>;
+
+        #[axum::async_trait]
+        impl super::DaphneService for Channel {
+            fn server_metrics(&self) -> &dyn daphne_service_utils::metrics::DaphneServiceMetrics {
+                // These tests don't care about metrics so we just store a static instance here so I
+                // can implement the DaphneService trait for Channel.
+                static METRICS: OnceLock<DaphnePromServiceMetrics> = OnceLock::new();
+                METRICS.get_or_init(|| {
+                    DaphnePromServiceMetrics::register(prometheus::default_registry()).unwrap()
+                })
+            }
+
+            fn signing_key(&self) -> Option<&p256::ecdsa::SigningKey> {
+                None
+            }
+
+            async fn check_bearer_token(
+                &self,
+                token: &BearerToken,
+                _sender: DapSender,
+                _task_id: TaskId,
+                _is_taskprov: bool,
+            ) -> Result<(), Either<String, DapError>> {
+                (token.as_str() == BEARER_TOKEN)
+                    .then_some(())
+                    .ok_or_else(|| Left("invalid token".into()))
+            }
+
+            async fn is_using_taskprov(&self, req: &DapRequestMeta) -> Result<bool, DapError> {
+                Ok(req.taskprov.is_some())
+            }
+        }
+
+        // setup a channel to "smuggle" the parsed request out of a handler
+        let (tx, mut rx) = mpsc::channel(1);
+
+        // create a router that takes the send end of the channel as state
+        let router = Router::new()
+            .route("/:version/:task_id/auth", get(auth_handler))
+            .route("/:version/:task_id/parse-mandatory-fields", get(handler))
+            .route(
+                "/:version/:task_id/:agg_job_id/parse-agg-job-id",
+                get(handler),
+            )
+            .route(
+                "/:version/:task_id/:collect_job_id/parse-collect-job-id",
+                get(handler),
+            )
+            .with_state(Arc::new(tx));
+
+        // unauthenticated handler that simply sends the received request through the channel
+        async fn handler(
+            State(ch): State<Arc<Channel>>,
+            req: UnauthenticatedDapRequestExtractor,
+        ) -> impl IntoResponse {
+            ch.send(req.0).await.unwrap();
+        }
+
+        // unauthenticated handler that simply sends the received request through the channel
+        async fn auth_handler(
+            State(ch): State<Arc<Channel>>,
+            req: DapRequestExtractor,
+        ) -> impl IntoResponse {
+            ch.send(req.0).await.unwrap();
+        }
+
+        move |req| {
+            Box::pin(async move {
+                let resp = match timeout(Duration::from_secs(1), router.oneshot(req))
+                    .await
+                    .unwrap()
+                {
+                    Ok(resp) => resp,
+                    Err(i) => match i {},
+                };
+
+                match resp.status() {
+                    StatusCode::NOT_FOUND => panic!("unsuported uri"),
+                    // get the request sent through the channel in the handler
+                    StatusCode::OK => Ok(rx.recv().now_or_never().unwrap().unwrap()),
+                    code => Err(code),
+                }
+            })
+        }
+    }
+
+    fn mk_task_id() -> TaskId {
+        TaskId(thread_rng().gen())
+    }
+
+    async fn parse_mandatory_fields(version: DapVersion) {
+        let test = test_router();
+
+        let task_id = mk_task_id();
+        let req = test(
+            Request::builder()
+                .uri(format!(
+                    "/{version}/{}/parse-mandatory-fields",
+                    task_id.to_base64url()
+                ))
+                .header(
+                    CONTENT_TYPE,
+                    DapMediaType::AggregateShareReq
+                        .as_str_for_version(version)
+                        .unwrap(),
+                )
+                .header(http_headers::DAP_AUTH_TOKEN, BEARER_TOKEN)
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(req.version, version);
+        assert_eq!(req.task_id, task_id);
+    }
+
+    async_test_versions! { parse_mandatory_fields }
+
+    async fn parse_agg_job_id(version: DapVersion) {
+        let test = test_router();
+
+        let task_id = mk_task_id();
+        let agg_job_id = AggregationJobId(thread_rng().gen());
+
+        let req = test(
+            Request::builder()
+                .uri(format!(
+                    "/{version}/{}/{}/parse-agg-job-id",
+                    task_id.to_base64url(),
+                    agg_job_id.to_base64url(),
+                ))
+                .header(
+                    CONTENT_TYPE,
+                    DapMediaType::AggregationJobInitReq
+                        .as_str_for_version(version)
+                        .unwrap(),
+                )
+                .header(http_headers::DAP_AUTH_TOKEN, BEARER_TOKEN)
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(req.resource, DapResource::AggregationJob(agg_job_id));
+        assert_eq!(req.task_id, task_id);
+    }
+
+    async_test_versions! { parse_agg_job_id }
+
+    async fn parse_collect_job_id(version: DapVersion) {
+        let test = test_router();
+
+        let task_id = mk_task_id();
+        let collect_job_id = CollectionJobId(thread_rng().gen());
+
+        let req = test(
+            Request::builder()
+                .uri(format!(
+                    "/{version}/{}/{}/parse-collect-job-id",
+                    task_id.to_base64url(),
+                    collect_job_id.to_base64url(),
+                ))
+                .header(
+                    CONTENT_TYPE,
+                    DapMediaType::CollectionReq
+                        .as_str_for_version(version)
+                        .unwrap(),
+                )
+                .header(http_headers::DAP_AUTH_TOKEN, BEARER_TOKEN)
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(req.resource, DapResource::CollectionJob(collect_job_id));
+        assert_eq!(req.task_id, task_id);
+    }
+
+    async_test_versions! { parse_collect_job_id }
+
+    async fn incorrect_bearer_tokens_are_rejected(version: DapVersion) {
+        let test = test_router();
+
+        let status_code = test(
+            Request::builder()
+                .uri(format!("/{version}/{}/auth", mk_task_id().to_base64url()))
+                .header(
+                    CONTENT_TYPE,
+                    DapMediaType::Report.as_str_for_version(version).unwrap(),
+                )
+                .header(http_headers::DAP_AUTH_TOKEN, "something incorrect")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap_err();
+
+        assert_eq!(status_code, StatusCode::UNAUTHORIZED);
+    }
+
+    async_test_versions! { incorrect_bearer_tokens_are_rejected }
+
+    async fn missing_auth_is_rejected(version: DapVersion) {
+        let test = test_router();
+
+        let status_code = test(
+            Request::builder()
+                .uri(format!("/{version}/{}/auth", mk_task_id().to_base64url()))
+                .header(
+                    CONTENT_TYPE,
+                    DapMediaType::Report.as_str_for_version(version).unwrap(),
+                )
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap_err();
+
+        assert_eq!(status_code, StatusCode::UNAUTHORIZED);
+    }
+
+    async_test_versions! { missing_auth_is_rejected }
+
+    async fn mtls_auth_is_enough(version: DapVersion) {
+        let test = test_router();
+
+        let req = test(
+            Request::builder()
+                .uri(format!("/{version}/{}/auth", mk_task_id().to_base64url()))
+                .header(
+                    CONTENT_TYPE,
+                    DapMediaType::Report.as_str_for_version(version).unwrap(),
+                )
+                .header("X-Client-Cert-Verified", "SUCCESS")
+                .header(http_headers::DAP_TASKPROV, "some-taskprov-string")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await;
+
+        req.unwrap();
+    }
+
+    async_test_versions! { mtls_auth_is_enough }
+
+    async fn incorrect_bearer_tokens_are_rejected_even_with_mtls_auth(version: DapVersion) {
+        let test = test_router();
+
+        let code = test(
+            Request::builder()
+                .uri(format!("/{version}/{}/auth", mk_task_id().to_base64url()))
+                .header(
+                    CONTENT_TYPE,
+                    DapMediaType::Report.as_str_for_version(version).unwrap(),
+                )
+                .header(http_headers::DAP_AUTH_TOKEN, "something incorrect")
+                .header("X-Client-Cert-Verified", "SUCCESS")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap_err();
+
+        assert_eq!(code, StatusCode::UNAUTHORIZED);
+    }
+
+    async_test_versions! { incorrect_bearer_tokens_are_rejected_even_with_mtls_auth }
+
+    async fn invalid_mtls_auth_is_rejected_despite_correct_bearer_token(version: DapVersion) {
+        let test = test_router();
+
+        let code = test(
+            Request::builder()
+                .uri(format!("/{version}/{}/auth", mk_task_id().to_base64url()))
+                .header(
+                    CONTENT_TYPE,
+                    DapMediaType::Report.as_str_for_version(version).unwrap(),
+                )
+                .header(http_headers::DAP_AUTH_TOKEN, BEARER_TOKEN)
+                .header("X-Client-Cert-Verified", "FAILED")
+                .body(Body::empty())
+                .unwrap(),
+        )
+        .await
+        .unwrap_err();
+
+        assert_eq!(code, StatusCode::UNAUTHORIZED);
+    }
+
+    async_test_versions! { invalid_mtls_auth_is_rejected_despite_correct_bearer_token }
+}
diff --git a/crates/daphne-server/src/router/helper.rs b/crates/daphne-server/src/router/helper.rs
index 029b3e9..8dc0937 100644
--- a/crates/daphne-server/src/router/helper.rs
+++ b/crates/daphne-server/src/router/helper.rs
@@ -9,8 +9,7 @@ use axum::{
     routing::{post, put},
 };
 use daphne::{
-    constants::DapMediaType,
-    error::DapAbort,
+    messages::{AggregateShareReq, AggregationJobInitReq},
     roles::{helper, DapHelper},
 };
 use http::StatusCode;
@@ -43,27 +42,15 @@ where
 )]
 async fn agg_job(
     State(app): State<Arc<App>>,
-    DapRequestExtractor(req): DapRequestExtractor,
+    DapRequestExtractor(req): DapRequestExtractor<AggregationJobInitReq>,
 ) -> AxumDapResponse {
-    match req.media_type {
-        Some(DapMediaType::AggregationJobInitReq) => {
-            let resp = helper::handle_agg_job_init_req(
-                &*app,
-                &req,
-                fetch_replay_protection_override(app.kv()).await,
-            )
-            .await;
-            AxumDapResponse::from_result_with_success_code(
-                resp,
-                app.server_metrics(),
-                StatusCode::CREATED,
-            )
-        }
-        m => AxumDapResponse::new_error(
-            DapAbort::BadRequest(format!("unexpected media type: {m:?}")),
-            app.server_metrics(),
-        ),
-    }
+    let resp = helper::handle_agg_job_init_req(
+        &*app,
+        req,
+        fetch_replay_protection_override(app.kv()).await,
+    )
+    .await;
+    AxumDapResponse::from_result_with_success_code(resp, app.server_metrics(), StatusCode::CREATED)
 }
 
 #[tracing::instrument(
@@ -76,13 +63,13 @@ async fn agg_job(
 )]
 async fn agg_share<A>(
     State(app): State<Arc<A>>,
-    DapRequestExtractor(req): DapRequestExtractor,
+    DapRequestExtractor(req): DapRequestExtractor<AggregateShareReq>,
 ) -> AxumDapResponse
 where
     A: DapHelper + DaphneService + Send + Sync,
 {
     AxumDapResponse::from_result(
-        helper::handle_agg_share_req(&*app, &req).await,
+        helper::handle_agg_share_req(&*app, req).await,
         app.server_metrics(),
     )
 }
diff --git a/crates/daphne-server/src/router/leader.rs b/crates/daphne-server/src/router/leader.rs
index db7fa22..2fa2e45 100644
--- a/crates/daphne-server/src/router/leader.rs
+++ b/crates/daphne-server/src/router/leader.rs
@@ -13,6 +13,7 @@ use axum::{
 use daphne::{
     constants::DapMediaType,
     error::DapAbort,
+    messages,
     roles::leader::{self, DapLeader},
     DapError, DapVersion,
 };
@@ -50,12 +51,12 @@ where
 )]
 async fn upload<A>(
     State(app): State<Arc<A>>,
-    UnauthenticatedDapRequestExtractor(req): UnauthenticatedDapRequestExtractor,
+    UnauthenticatedDapRequestExtractor(req): UnauthenticatedDapRequestExtractor<messages::Report>,
 ) -> Response
 where
     A: DapLeader + DaphneService + Send + Sync,
 {
-    match leader::handle_upload_req(&*app, &req).await {
+    match leader::handle_upload_req(&*app, req).await {
         Ok(()) => StatusCode::OK.into_response(),
         Err(e) => AxumDapResponse::new_error(e, app.server_metrics()).into_response(),
     }
@@ -70,7 +71,7 @@ where
 )]
 async fn get_collect_uri<A>(
     State(app): State<Arc<A>>,
-    DapRequestExtractor(req): DapRequestExtractor,
+    DapRequestExtractor(req): DapRequestExtractor<messages::CollectionReq>,
 ) -> Response
 where
     A: DapLeader + DaphneService + Send + Sync,
@@ -92,7 +93,7 @@ where
 )]
 async fn collect<A>(
     State(app): State<Arc<A>>,
-    DapRequestExtractor(req): DapRequestExtractor,
+    DapRequestExtractor(req): DapRequestExtractor<()>,
 ) -> Response
 where
     A: DapLeader + DaphneService + Send + Sync,
diff --git a/crates/daphne-server/src/router/mod.rs b/crates/daphne-server/src/router/mod.rs
index 38bbce5..ad7b716 100644
--- a/crates/daphne-server/src/router/mod.rs
+++ b/crates/daphne-server/src/router/mod.rs
@@ -2,6 +2,7 @@
 // SPDX-License-Identifier: BSD-3-Clause
 
 mod aggregator;
+mod extractor;
 mod helper;
 mod leader;
 #[cfg(feature = "test-utils")]
@@ -10,37 +11,28 @@ pub mod test_routes;
 use std::sync::Arc;
 
 use axum::{
-    async_trait,
     body::HttpBody,
-    extract::{FromRequest, FromRequestParts, Path, State},
+    extract::State,
     http::{header::CONTENT_TYPE, HeaderValue, StatusCode},
     middleware::Next,
     response::IntoResponse,
     Json,
 };
 use daphne::{
-    constants::DapMediaType,
-    error::DapAbort,
-    fatal_error,
-    messages::{AggregationJobId, CollectionJobId, TaskId},
-    DapError, DapRequest, DapResource, DapResponse, DapSender, DapVersion,
-};
-use daphne_service_utils::{
-    bearer_token::BearerToken,
-    http_headers,
-    metrics::{self, DaphneServiceMetrics},
-    DapRole,
+    error::DapAbort, fatal_error, messages::TaskId, DapError, DapRequestMeta, DapResponse,
+    DapSender,
 };
+use daphne_service_utils::{bearer_token::BearerToken, metrics::DaphneServiceMetrics, DapRole};
 use either::Either;
-use http::{HeaderMap, Request};
-use serde::Deserialize;
+use http::Request;
 
 use crate::App;
+use extractor::{DapRequestExtractor, UnauthenticatedDapRequestExtractor};
 
 type Router<A, B> = axum::Router<Arc<A>, B>;
 
 /// Capabilities necessary when running a native daphne service.
-#[async_trait]
+#[axum::async_trait]
 pub trait DaphneService {
     /// The service metrics
     fn server_metrics(&self) -> &dyn DaphneServiceMetrics;
@@ -65,10 +57,10 @@ pub trait DaphneService {
     ) -> Result<(), Either<String, DapError>>;
 
     /// Checks if this request intends to use taskprov.
-    async fn is_using_taskprov(&self, req: &DapRequest) -> Result<bool, DapError>;
+    async fn is_using_taskprov(&self, req: &DapRequestMeta) -> Result<bool, DapError>;
 }
 
-#[async_trait]
+#[axum::async_trait]
 impl<S> DaphneService for Arc<S>
 where
     S: DaphneService + Send + Sync,
@@ -91,7 +83,7 @@ where
         S::check_bearer_token(&**self, presented_token, sender, task_id, is_taskprov).await
     }
 
-    async fn is_using_taskprov(&self, req: &DapRequest) -> Result<bool, DapError> {
+    async fn is_using_taskprov(&self, req: &DapRequestMeta) -> Result<bool, DapError> {
         S::is_using_taskprov(&**self, req).await
     }
 }
@@ -187,21 +179,22 @@ impl AxumDapResponse {
             DapError::Fatal(e) => Err(e),
             DapError::Abort(abort) => Ok(abort),
         };
-        let status = if let Err(_e) = &error {
-            // TODO(mendess) uncomment the line below
-            // self.error_reporter.report_abort(&e);
-            StatusCode::INTERNAL_SERVER_ERROR
-        } else {
-            StatusCode::BAD_REQUEST
-        };
-        let problem_details = match error {
-            Ok(error) => {
-                tracing::error!(?error, "request aborted due to protocol abort");
-                error.into_problem_details()
+        let (status, problem_details) = match error {
+            Ok(abort) => {
+                tracing::error!(error = ?abort, "request aborted due to protocol abort");
+                let status = if let DapAbort::UnauthorizedRequest { .. } = abort {
+                    StatusCode::UNAUTHORIZED
+                } else {
+                    StatusCode::BAD_REQUEST
+                };
+                (status, abort.into_problem_details())
             }
-            Err(error) => {
-                tracing::error!(?error, "request aborted due to fatal error");
-                DapError::Fatal(error).into_problem_details()
+            Err(fatal_error) => {
+                tracing::error!(error = ?fatal_error, "request aborted due to fatal error");
+                // TODO(mendess) uncomment the line below
+                // self.error_reporter.report_abort(&e);
+                let problem_details = fatal_error.into_problem_details();
+                (StatusCode::INTERNAL_SERVER_ERROR, problem_details)
             }
         };
         // this to string is bounded by the
@@ -242,510 +235,3 @@ impl IntoResponse for AxumDapResponse {
         self.0
     }
 }
-
-/// An axum extractor capable of parsing a [`DapRequest`].
-#[derive(Debug)]
-struct UnauthenticatedDapRequestExtractor(pub DapRequest);
-
-#[async_trait]
-impl<S, B> FromRequest<S, B> for UnauthenticatedDapRequestExtractor
-where
-    S: DaphneService + Send + Sync,
-    B: HttpBody + Send + 'static,
-    <B as HttpBody>::Data: Send,
-{
-    type Rejection = (StatusCode, String);
-
-    async fn from_request(req: Request<B>, state: &S) -> Result<Self, Self::Rejection> {
-        #[derive(Debug, Deserialize)]
-        #[serde(deny_unknown_fields)]
-        struct PathParams {
-            version: DapVersion,
-            #[serde(default, with = "daphne::messages::base64url")]
-            task_id: TaskId,
-            #[serde(default, with = "daphne::messages::base64url_option")]
-            agg_job_id: Option<AggregationJobId>,
-            #[serde(default, with = "daphne::messages::base64url_option")]
-            collect_job_id: Option<CollectionJobId>,
-        }
-
-        let (mut parts, body) = req.into_parts();
-        let Path(PathParams {
-            version,
-            task_id,
-            agg_job_id,
-            collect_job_id,
-        }) = Path::from_request_parts(&mut parts, state)
-            .await
-            .map_err(|e| (StatusCode::BAD_REQUEST, e.to_string()))?;
-
-        let media_type = if let Some(content_type) = parts.headers.get(CONTENT_TYPE) {
-            let content_type = content_type.to_str().map_err(|_| {
-                let msg = "header value contains non ascii or invisible characters".into();
-                (StatusCode::BAD_REQUEST, msg)
-            })?;
-            let mt = DapMediaType::from_str_for_version(version, content_type)
-                .ok_or_else(|| (StatusCode::BAD_REQUEST, "invalid media type".into()))?;
-            Some(mt)
-        } else {
-            None
-        };
-
-        let taskprov = extract_header_as_string(&parts.headers, http_headers::DAP_TASKPROV);
-
-        // TODO(mendess): this is very eager, we could redesign DapResponse later to allow for
-        // streaming of data.
-        let payload = hyper::body::to_bytes(body).await;
-
-        let Ok(payload) = payload else {
-            return Err((
-                StatusCode::INTERNAL_SERVER_ERROR,
-                "failed to get payload".into(),
-            ));
-        };
-
-        let (task_id, resource) = {
-            let resource = match media_type {
-                Some(DapMediaType::AggregationJobInitReq) => {
-                    if let Some(agg_job_id) = agg_job_id {
-                        DapResource::AggregationJob(agg_job_id)
-                    } else {
-                        // Missing or invalid agg job ID. This should be handled as a bad
-                        // request (undefined resource) by the caller.
-                        DapResource::Undefined
-                    }
-                }
-                Some(DapMediaType::CollectReq) => {
-                    if let Some(collect_job_id) = collect_job_id {
-                        DapResource::CollectionJob(collect_job_id)
-                    } else {
-                        // Missing or invalid agg job ID. This should be handled as a bad
-                        // request (undefined resource) by the caller.
-                        DapResource::Undefined
-                    }
-                }
-                _ => DapResource::Undefined,
-            };
-
-            (task_id, resource)
-        };
-
-        let request = DapRequest {
-            version,
-            task_id,
-            resource,
-            payload: payload.to_vec(),
-            media_type,
-            taskprov,
-        };
-
-        Ok(UnauthenticatedDapRequestExtractor(request))
-    }
-}
-
-/// An axum extractor capable of parsing a [`DapRequest`].
-///
-/// This extractor asserts that the request is authenticated.
-#[derive(Debug)]
-struct DapRequestExtractor(pub DapRequest);
-
-#[async_trait]
-impl<S, B> FromRequest<S, B> for DapRequestExtractor
-where
-    S: DaphneService + Send + Sync,
-    B: HttpBody + Send + 'static,
-    <B as HttpBody>::Data: Send,
-{
-    type Rejection = (StatusCode, String);
-
-    async fn from_request(req: Request<B>, state: &S) -> Result<Self, Self::Rejection> {
-        let bearer_token = extract_header_as_string(req.headers(), http_headers::DAP_AUTH_TOKEN)
-            .map(BearerToken::from);
-        let cf_tls_client_auth = extract_header_as_string(req.headers(), "X-Client-Cert-Verified");
-
-        let request = UnauthenticatedDapRequestExtractor::from_request(req, state)
-            .await?
-            .0;
-
-        let is_taskprov = state
-            .is_using_taskprov(&request)
-            .await
-            .map_err(|_| (StatusCode::INTERNAL_SERVER_ERROR, "internal error".into()))?;
-
-        let bearer_authed = if let Some((token, sender)) =
-            bearer_token.zip(request.media_type.map(|m| m.sender()))
-        {
-            state
-                .server_metrics()
-                .auth_method_inc(metrics::AuthMethod::BearerToken);
-            state
-                .check_bearer_token(&token, sender, request.task_id, is_taskprov)
-                .await
-                .map_err(|reason| {
-                    reason.either(
-                        |reason| (StatusCode::UNAUTHORIZED, reason),
-                        |_| (StatusCode::INTERNAL_SERVER_ERROR, "internal error".into()),
-                    )
-                })?;
-            true
-        } else {
-            false
-        };
-        let mtls_authed = if let Some(verification_result) = cf_tls_client_auth {
-            state
-                .server_metrics()
-                .auth_method_inc(metrics::AuthMethod::TlsClientAuth);
-            // we always check if mtls succedded even if ...
-            if verification_result != "SUCCESS" {
-                return Err((
-                    StatusCode::UNAUTHORIZED,
-                    format!("Invalid TLS certificate ({verification_result})"),
-                ));
-            }
-            // ... we only allow mtls auth for taskprov tasks
-            is_taskprov
-        } else {
-            false
-        };
-
-        if bearer_authed || mtls_authed {
-            Ok(Self(request))
-        } else {
-            Err((
-                StatusCode::UNAUTHORIZED,
-                "No suitable authorization method was found".into(),
-            ))
-        }
-    }
-}
-
-fn extract_header_as_string(headers: &HeaderMap, header: &'static str) -> Option<String> {
-    headers.get(header)?.to_str().ok().map(ToString::to_string)
-}
-
-#[cfg(test)]
-mod test {
-    use std::{
-        sync::{Arc, OnceLock},
-        time::Duration,
-    };
-
-    use axum::{
-        body::{Body, HttpBody},
-        extract::State,
-        http::{header::CONTENT_TYPE, Request, StatusCode},
-        response::IntoResponse,
-        routing::get,
-        Router,
-    };
-    use daphne::{
-        async_test_versions,
-        messages::{AggregationJobId, Base64Encode, CollectionJobId, TaskId},
-        DapError, DapRequest, DapResource, DapSender, DapVersion,
-    };
-    use daphne_service_utils::{
-        bearer_token::BearerToken, http_headers, metrics::DaphnePromServiceMetrics,
-    };
-    use either::Either::{self, Left};
-    use futures::{future::BoxFuture, FutureExt};
-    use rand::{thread_rng, Rng};
-    use tokio::{
-        sync::mpsc::{self, Sender},
-        time::timeout,
-    };
-    use tower::ServiceExt;
-
-    use crate::router::DapRequestExtractor;
-
-    use super::UnauthenticatedDapRequestExtractor;
-
-    const BEARER_TOKEN: &str = "test-token";
-
-    /// Return a function that will parse a request using the [`DapRequestExtractor`] or
-    /// [`UnauthenticatedDapRequestExtractor`] and return the parsed request.
-    ///
-    /// The possible request URIs that are supported by this parser are:
-    ///  - `/:version/:task_id/auth` uses the [`DapRequestExtractor`]
-    ///  - `/:version/:task_id/parse-mandatory-fields` uses the [`UnauthenticatedDapRequestExtractor`]
-    ///  - `/:version/:agg_job_id/parse-agg-job-id` uses the [`UnauthenticatedDapRequestExtractor`]
-    ///  - `/:version/:collect_job_id/parse-collect-job-id` uses the [`UnauthenticatedDapRequestExtractor`]
-    fn test_router<B>(
-    ) -> impl FnOnce(Request<B>) -> BoxFuture<'static, Result<DapRequest, StatusCode>>
-    where
-        B: Send + Sync + 'static + HttpBody,
-        B::Data: Send,
-        B::Error: Send + Sync + std::error::Error,
-    {
-        type Channel = Sender<DapRequest>;
-
-        #[axum::async_trait]
-        impl super::DaphneService for Channel {
-            fn server_metrics(&self) -> &dyn daphne_service_utils::metrics::DaphneServiceMetrics {
-                // These tests don't care about metrics so we just store a static instance here so I
-                // can implement the DaphneService trait for Channel.
-                static METRICS: OnceLock<DaphnePromServiceMetrics> = OnceLock::new();
-                METRICS.get_or_init(|| {
-                    DaphnePromServiceMetrics::register(prometheus::default_registry()).unwrap()
-                })
-            }
-
-            fn signing_key(&self) -> Option<&p256::ecdsa::SigningKey> {
-                None
-            }
-
-            async fn check_bearer_token(
-                &self,
-                token: &BearerToken,
-                _sender: DapSender,
-                _task_id: TaskId,
-                _is_taskprov: bool,
-            ) -> Result<(), Either<String, DapError>> {
-                (token.as_str() == BEARER_TOKEN)
-                    .then_some(())
-                    .ok_or_else(|| Left("invalid token".into()))
-            }
-
-            async fn is_using_taskprov(&self, req: &DapRequest) -> Result<bool, DapError> {
-                Ok(req.taskprov.is_some())
-            }
-        }
-
-        // setup a channel to "smuggle" the parsed request out of a handler
-        let (tx, mut rx) = mpsc::channel(1);
-
-        // create a router that takes the send end of the channel as state
-        let router = Router::new()
-            .route("/:version/:task_id/auth", get(auth_handler))
-            .route("/:version/:task_id/parse-mandatory-fields", get(handler))
-            .route(
-                "/:version/:task_id/:agg_job_id/parse-agg-job-id",
-                get(handler),
-            )
-            .route(
-                "/:version/:task_id/:collect_job_id/parse-collect-job-id",
-                get(handler),
-            )
-            .with_state(Arc::new(tx));
-
-        // unauthenticated handler that simply sends the received request through the channel
-        async fn handler(
-            State(ch): State<Arc<Channel>>,
-            UnauthenticatedDapRequestExtractor(req): UnauthenticatedDapRequestExtractor,
-        ) -> impl IntoResponse {
-            ch.send(req).await.unwrap();
-        }
-
-        // unauthenticated handler that simply sends the received request through the channel
-        async fn auth_handler(
-            State(ch): State<Arc<Channel>>,
-            DapRequestExtractor(req): DapRequestExtractor,
-        ) -> impl IntoResponse {
-            ch.send(req).await.unwrap();
-        }
-
-        move |req| {
-            Box::pin(async move {
-                let resp = match timeout(Duration::from_secs(1), router.oneshot(req))
-                    .await
-                    .unwrap()
-                {
-                    Ok(resp) => resp,
-                    Err(i) => match i {},
-                };
-
-                match resp.status() {
-                    StatusCode::NOT_FOUND => panic!("unsuported uri"),
-                    StatusCode::BAD_REQUEST => {
-                        panic!(
-                            "parsing failed: {}",
-                            String::from_utf8_lossy(
-                                &hyper::body::to_bytes(resp.into_body()).await.unwrap()
-                            )
-                        )
-                    }
-                    // get the request sent through the channel in the handler
-                    StatusCode::OK => Ok(rx.recv().now_or_never().unwrap().unwrap()),
-                    code => Err(code),
-                }
-            })
-        }
-    }
-
-    fn mk_task_id() -> TaskId {
-        TaskId(thread_rng().gen())
-    }
-
-    async fn parse_mandatory_fields(version: DapVersion) {
-        let test = test_router();
-
-        let task_id = mk_task_id();
-        let req = test(
-            Request::builder()
-                .uri(format!(
-                    "/{version}/{}/parse-mandatory-fields",
-                    task_id.to_base64url()
-                ))
-                .header(CONTENT_TYPE, "application/dap-aggregation-job-init-req")
-                .header(http_headers::DAP_AUTH_TOKEN, BEARER_TOKEN)
-                .body(Body::empty())
-                .unwrap(),
-        )
-        .await
-        .unwrap();
-
-        assert_eq!(req.version, version);
-        assert_eq!(req.task_id, task_id);
-    }
-
-    async_test_versions! { parse_mandatory_fields }
-
-    async fn parse_agg_job_id(version: DapVersion) {
-        let test = test_router();
-
-        let task_id = mk_task_id();
-        let agg_job_id = AggregationJobId(thread_rng().gen());
-
-        let req = test(
-            Request::builder()
-                .uri(format!(
-                    "/{version}/{}/{}/parse-agg-job-id",
-                    task_id.to_base64url(),
-                    agg_job_id.to_base64url(),
-                ))
-                .header(CONTENT_TYPE, "application/dap-aggregation-job-init-req")
-                .header(http_headers::DAP_AUTH_TOKEN, BEARER_TOKEN)
-                .body(Body::empty())
-                .unwrap(),
-        )
-        .await
-        .unwrap();
-
-        assert_eq!(req.resource, DapResource::AggregationJob(agg_job_id));
-        assert_eq!(req.task_id, task_id);
-    }
-
-    async_test_versions! { parse_agg_job_id }
-
-    async fn parse_collect_job_id(version: DapVersion) {
-        let test = test_router();
-
-        let task_id = mk_task_id();
-        let collect_job_id = CollectionJobId(thread_rng().gen());
-
-        let req = test(
-            Request::builder()
-                .uri(format!(
-                    "/{version}/{}/{}/parse-collect-job-id",
-                    task_id.to_base64url(),
-                    collect_job_id.to_base64url(),
-                ))
-                .header(CONTENT_TYPE, "application/dap-collect-req")
-                .header(http_headers::DAP_AUTH_TOKEN, BEARER_TOKEN)
-                .body(Body::empty())
-                .unwrap(),
-        )
-        .await
-        .unwrap();
-
-        assert_eq!(req.resource, DapResource::CollectionJob(collect_job_id));
-        assert_eq!(req.task_id, task_id);
-    }
-
-    async_test_versions! { parse_collect_job_id }
-
-    async fn incorrect_bearer_tokens_are_rejected(version: DapVersion) {
-        let test = test_router();
-
-        let status_code = test(
-            Request::builder()
-                .uri(format!("/{version}/{}/auth", mk_task_id().to_base64url()))
-                .header(CONTENT_TYPE, "application/dap-aggregation-job-init-req")
-                .header(http_headers::DAP_AUTH_TOKEN, "something incorrect")
-                .body(Body::empty())
-                .unwrap(),
-        )
-        .await
-        .unwrap_err();
-
-        assert_eq!(status_code, StatusCode::UNAUTHORIZED);
-    }
-
-    async_test_versions! { incorrect_bearer_tokens_are_rejected }
-
-    async fn missing_auth_is_rejected(version: DapVersion) {
-        let test = test_router();
-
-        let status_code = test(
-            Request::builder()
-                .uri(format!("/{version}/{}/auth", mk_task_id().to_base64url()))
-                .header(CONTENT_TYPE, "application/dap-aggregation-job-init-req")
-                .body(Body::empty())
-                .unwrap(),
-        )
-        .await
-        .unwrap_err();
-
-        assert_eq!(status_code, StatusCode::UNAUTHORIZED);
-    }
-
-    async_test_versions! { missing_auth_is_rejected }
-
-    async fn mtls_auth_is_enough(version: DapVersion) {
-        let test = test_router();
-
-        let req = test(
-            Request::builder()
-                .uri(format!("/{version}/{}/auth", mk_task_id().to_base64url()))
-                .header(CONTENT_TYPE, "application/dap-aggregation-job-init-req")
-                .header("X-Client-Cert-Verified", "SUCCESS")
-                .header(http_headers::DAP_TASKPROV, "some-taskprov-string")
-                .body(Body::empty())
-                .unwrap(),
-        )
-        .await;
-
-        req.unwrap();
-    }
-
-    async_test_versions! { mtls_auth_is_enough }
-
-    async fn incorrect_bearer_tokens_are_rejected_even_with_mtls_auth(version: DapVersion) {
-        let test = test_router();
-
-        let code = test(
-            Request::builder()
-                .uri(format!("/{version}/{}/auth", mk_task_id().to_base64url()))
-                .header(CONTENT_TYPE, "application/dap-aggregation-job-init-req")
-                .header(http_headers::DAP_AUTH_TOKEN, "something incorrect")
-                .header("X-Client-Cert-Verified", "SUCCESS")
-                .body(Body::empty())
-                .unwrap(),
-        )
-        .await
-        .unwrap_err();
-
-        assert_eq!(code, StatusCode::UNAUTHORIZED);
-    }
-
-    async_test_versions! { incorrect_bearer_tokens_are_rejected_even_with_mtls_auth }
-
-    async fn invalid_mtls_auth_is_rejected_despite_correct_bearer_token(version: DapVersion) {
-        let test = test_router();
-
-        let code = test(
-            Request::builder()
-                .uri(format!("/{version}/{}/auth", mk_task_id().to_base64url()))
-                .header(CONTENT_TYPE, "application/dap-aggregation-job-init-req")
-                .header(http_headers::DAP_AUTH_TOKEN, BEARER_TOKEN)
-                .header("X-Client-Cert-Verified", "FAILED")
-                .body(Body::empty())
-                .unwrap(),
-        )
-        .await
-        .unwrap_err();
-
-        assert_eq!(code, StatusCode::UNAUTHORIZED);
-    }
-
-    async_test_versions! { invalid_mtls_auth_is_rejected_despite_correct_bearer_token }
-}
diff --git a/crates/daphne/src/constants.rs b/crates/daphne/src/constants.rs
index e9d2f42..589e9ad 100644
--- a/crates/daphne/src/constants.rs
+++ b/crates/daphne/src/constants.rs
@@ -23,7 +23,7 @@ pub enum DapMediaType {
     AggregationJobResp,
     AggregateShareReq,
     AggregateShare,
-    CollectReq,
+    CollectionReq,
     Collection,
     HpkeConfigList,
     Report,
@@ -40,7 +40,7 @@ impl DapMediaType {
             | Self::HpkeConfigList => DapSender::Leader,
             Self::AggregationJobResp | Self::AggregateShare => DapSender::Helper,
             Self::Report => DapSender::Client,
-            Self::CollectReq => DapSender::Collector,
+            Self::CollectionReq => DapSender::Collector,
         }
     }
 
@@ -54,7 +54,7 @@ impl DapMediaType {
             MEDIA_TYPE_COLLECTION => Self::Collection,
             MEDIA_TYPE_HPKE_CONFIG_LIST => Self::HpkeConfigList,
             MEDIA_TYPE_AGG_SHARE_REQ => Self::AggregateShareReq,
-            MEDIA_TYPE_COLLECT_REQ => Self::CollectReq,
+            MEDIA_TYPE_COLLECT_REQ => Self::CollectionReq,
             MEDIA_TYPE_REPORT => Self::Report,
             _ => return None,
         };
@@ -69,7 +69,7 @@ impl DapMediaType {
             Self::AggregationJobResp => Some(MEDIA_TYPE_AGG_JOB_RESP),
             Self::AggregateShareReq => Some(MEDIA_TYPE_AGG_SHARE_REQ),
             Self::AggregateShare => Some(MEDIA_TYPE_AGG_SHARE),
-            Self::CollectReq => Some(MEDIA_TYPE_COLLECT_REQ),
+            Self::CollectionReq => Some(MEDIA_TYPE_COLLECT_REQ),
             Self::Collection => Some(MEDIA_TYPE_COLLECTION),
             Self::HpkeConfigList => Some(MEDIA_TYPE_HPKE_CONFIG_LIST),
             Self::Report => Some(MEDIA_TYPE_REPORT),
@@ -122,7 +122,7 @@ mod test {
         );
         assert_eq!(
             DapMediaType::from_str_for_version(DapVersion::Draft09, "application/dap-collect-req"),
-            Some(DapMediaType::CollectReq),
+            Some(DapMediaType::CollectionReq),
         );
         assert_eq!(
             DapMediaType::from_str_for_version(DapVersion::Draft09, "application/dap-collection"),
diff --git a/crates/daphne/src/error/aborts.rs b/crates/daphne/src/error/aborts.rs
index f653fbd..99ca791 100644
--- a/crates/daphne/src/error/aborts.rs
+++ b/crates/daphne/src/error/aborts.rs
@@ -4,9 +4,10 @@
 //! Definitions and tooling for DAP protocol aborts.
 
 use crate::{
+    constants::DapMediaType,
     fatal_error,
     messages::{AggregationJobId, ReportId, TaskId, TransitionFailure},
-    DapError, DapMediaType, DapRequest, DapVersion,
+    DapError, DapRequestMeta, DapVersion,
 };
 use hex::FromHexError;
 use prio::codec::CodecError;
@@ -181,7 +182,7 @@ impl DapAbort {
     }
 
     /// Abort due to unexpected value for HTTP content-type header.
-    pub fn content_type(req: &DapRequest, expected: DapMediaType) -> Self {
+    pub fn content_type(req: &DapRequestMeta, expected: DapMediaType) -> Self {
         let want_content_type = expected.as_str_for_version(req.version).unwrap_or_else(|| {
             unreachable!("unexpected content-type for DAP version {:?}", req.version)
         });
diff --git a/crates/daphne/src/error/mod.rs b/crates/daphne/src/error/mod.rs
index c510a14..0d7383a 100644
--- a/crates/daphne/src/error/mod.rs
+++ b/crates/daphne/src/error/mod.rs
@@ -32,21 +32,6 @@ pub enum DapError {
 }
 
 impl DapError {
-    pub fn into_problem_details(self) -> ProblemDetails {
-        if let Self::Abort(a) = self {
-            return a.into_problem_details();
-        }
-
-        ProblemDetails {
-            typ: None,
-            title: "Internal server error".into(),
-            agg_job_id: None,
-            task_id: None,
-            instance: "/problem-details/internal-server-error".into(),
-            detail: None,
-        }
-    }
-
     /// Construct a fatal encoding error.
     pub fn encoding(e: CodecError) -> DapError {
         DapError::Fatal(FatalDapError(format!(
@@ -68,6 +53,19 @@ impl DapError {
 #[derive(Clone, PartialEq, Eq)]
 pub struct FatalDapError(pub(crate) String);
 
+impl FatalDapError {
+    pub fn into_problem_details(self) -> ProblemDetails {
+        ProblemDetails {
+            typ: None,
+            title: "Internal server error".into(),
+            agg_job_id: None,
+            task_id: None,
+            instance: "/problem-details/internal-server-error".into(),
+            detail: None,
+        }
+    }
+}
+
 impl std::error::Error for FatalDapError {}
 
 impl Display for FatalDapError {
diff --git a/crates/daphne/src/lib.rs b/crates/daphne/src/lib.rs
index 3b361f6..33b38e9 100644
--- a/crates/daphne/src/lib.rs
+++ b/crates/daphne/src/lib.rs
@@ -58,14 +58,13 @@ use crate::{
     error::DapAbort,
     hpke::HpkeReceiverConfig,
     messages::{
-        AggregationJobId, BatchId, BatchSelector, Collection, CollectionJobId, Duration, Interval,
-        PartialBatchSelector, ReportId, TaskId, Time,
+        BatchId, BatchSelector, Collection, Duration, Interval, PartialBatchSelector, ReportId,
+        TaskId, Time,
     },
     vdaf::{
         Prio3Config, VdafAggregateShare, VdafConfig, VdafPrepShare, VdafPrepState, VdafVerifyKey,
     },
 };
-use constants::DapMediaType;
 pub use error::DapError;
 use error::FatalDapError;
 use hpke::{HpkeConfig, HpkeKemId};
@@ -91,6 +90,7 @@ use url::Url;
 #[cfg(feature = "experimental")]
 use vdaf::mastic::MasticWeight;
 
+pub use messages::request::{DapRequest, DapRequestMeta, DapResource, DapResponse};
 pub use protocol::aggregator::{
     EarlyReportState, EarlyReportStateConsumed, EarlyReportStateInitialized,
 };
@@ -1048,90 +1048,6 @@ pub enum DapSender {
     Leader,
 }
 
-/// Types of resources associated with DAP tasks.
-#[derive(Clone, Debug, Default, PartialEq, Eq)]
-pub enum DapResource {
-    /// Aggregation job resource.
-    AggregationJob(AggregationJobId),
-
-    /// Collection job resource.
-    CollectionJob(CollectionJobId),
-
-    /// Undefined (or undetermined) resource.
-    ///
-    /// The resource of a DAP request is undefined if there is not a unique object (in the context
-    /// of a DAP task) that the request pertains to. For example:
-    ///
-    ///   * The Client->Aggregator request for the HPKE config or to upload a report
-    ///   * The Leader->Helper request for an aggregate share
-    ///
-    /// The resource of a DAP request is undetermined if its identifier could not be parsed from
-    /// request path.
-    #[default]
-    Undefined,
-}
-
-/// DAP request.
-#[derive(Debug)]
-pub struct DapRequest {
-    /// Protocol version indicated by the request.
-    pub version: DapVersion,
-
-    /// Request media type, sent in the "content-type" header of the HTTP request.
-    pub media_type: Option<DapMediaType>,
-
-    /// ID of the task with which the request is associated.
-    pub task_id: TaskId,
-
-    /// The resource with which this request is associated.
-    pub resource: DapResource,
-
-    /// Request payload.
-    pub payload: Vec<u8>,
-
-    /// taskprov: The task advertisement, sent in the `dap-taskprov` header.
-    pub taskprov: Option<String>,
-}
-
-#[cfg(test)]
-impl Default for DapRequest {
-    fn default() -> Self {
-        Self {
-            version: DapVersion::Draft09,
-            media_type: None,
-            task_id: Default::default(),
-            resource: Default::default(),
-            payload: Default::default(),
-            taskprov: Default::default(),
-        }
-    }
-}
-
-impl DapRequest {
-    /// Return the collection job ID, handling a missing ID as a user error.
-    pub fn collection_job_id(&self) -> Result<&CollectionJobId, DapAbort> {
-        if let DapResource::CollectionJob(collection_job_id) = &self.resource {
-            Ok(collection_job_id)
-        } else {
-            Err(DapAbort::BadRequest(
-                "missing or malformed collection job ID".into(),
-            ))
-        }
-    }
-
-    pub fn sender(&self) -> Option<DapSender> {
-        self.media_type.map(|m| m.sender())
-    }
-}
-
-/// DAP response.
-#[derive(Debug)]
-pub struct DapResponse {
-    pub version: DapVersion,
-    pub media_type: DapMediaType,
-    pub payload: Vec<u8>,
-}
-
 /// Status of a collect job.
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "snake_case")]
diff --git a/crates/daphne/src/messages/mod.rs b/crates/daphne/src/messages/mod.rs
index cd89960..15328f4 100644
--- a/crates/daphne/src/messages/mod.rs
+++ b/crates/daphne/src/messages/mod.rs
@@ -3,6 +3,7 @@
 
 //! Messages in the DAP protocol.
 
+pub mod request;
 pub mod taskprov;
 
 use crate::{
diff --git a/crates/daphne/src/messages/request.rs b/crates/daphne/src/messages/request.rs
new file mode 100644
index 0000000..a698ef3
--- /dev/null
+++ b/crates/daphne/src/messages/request.rs
@@ -0,0 +1,111 @@
+// Copyright (c) 2024 Cloudflare, Inc. All rights reserved.
+// SPDX-License-Identifier: BSD-3-Clause
+
+use std::ops::Deref;
+
+use crate::{
+    constants::DapMediaType,
+    error::DapAbort,
+    messages::{AggregationJobId, CollectionJobId, TaskId},
+    DapSender, DapVersion,
+};
+
+/// Types of resources associated with DAP tasks.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub enum DapResource {
+    /// Aggregation job resource.
+    AggregationJob(AggregationJobId),
+
+    /// Collection job resource.
+    CollectionJob(CollectionJobId),
+
+    /// Undefined (or undetermined) resource.
+    ///
+    /// The resource of a DAP request is undefined if there is not a unique object (in the context
+    /// of a DAP task) that the request pertains to. For example:
+    ///
+    ///   * The Client->Aggregator request for the HPKE config or to upload a report
+    ///   * The Leader->Helper request for an aggregate share
+    ///
+    /// The resource of a DAP request is undetermined if its identifier could not be parsed from
+    /// request path.
+    #[default]
+    Undefined,
+}
+
+/// Fields common to all DAP requests.
+#[derive(Debug)]
+#[cfg_attr(test, derive(Default))]
+pub struct DapRequestMeta {
+    /// Protocol version indicated by the request.
+    pub version: DapVersion,
+
+    /// Request media type, sent in the "content-type" header of the HTTP request.
+    pub media_type: Option<DapMediaType>,
+
+    /// ID of the task with which the request is associated.
+    pub task_id: TaskId,
+
+    /// taskprov: The task advertisement, sent in the `dap-taskprov` header.
+    pub taskprov: Option<String>,
+
+    /// The resource with which this request is associated.
+    pub resource: DapResource,
+}
+
+impl DapRequestMeta {
+    /// Checks the request content type against the expected content type.
+    pub fn get_checked_media_type(&self, expected: DapMediaType) -> Result<DapMediaType, DapAbort> {
+        self.media_type
+            .filter(|m| *m == expected)
+            .ok_or_else(|| DapAbort::content_type(self, expected))
+    }
+}
+
+/// DAP request.
+#[derive(Debug)]
+#[cfg_attr(test, derive(Default))]
+pub struct DapRequest<P> {
+    pub meta: DapRequestMeta,
+
+    /// Request payload.
+    pub payload: P,
+}
+
+impl<P> AsRef<DapRequestMeta> for DapRequest<P> {
+    fn as_ref(&self) -> &DapRequestMeta {
+        &self.meta
+    }
+}
+
+impl<P> Deref for DapRequest<P> {
+    type Target = DapRequestMeta;
+    fn deref(&self) -> &Self::Target {
+        &self.meta
+    }
+}
+
+impl<P> DapRequest<P> {
+    /// Return the collection job ID, handling a missing ID as a user error.
+    pub fn collection_job_id(&self) -> Result<&CollectionJobId, DapAbort> {
+        if let DapResource::CollectionJob(collection_job_id) = &self.resource {
+            Ok(collection_job_id)
+        } else {
+            Err(DapAbort::BadRequest(
+                "missing or malformed collection job ID".into(),
+            ))
+        }
+    }
+
+    pub fn sender(&self) -> Option<DapSender> {
+        self.media_type.map(|m| m.sender())
+    }
+}
+
+/// DAP response.
+#[derive(Debug)]
+pub struct DapResponse {
+    pub version: DapVersion,
+    pub media_type: DapMediaType,
+    pub payload: Vec<u8>,
+}
diff --git a/crates/daphne/src/roles/aggregator.rs b/crates/daphne/src/roles/aggregator.rs
index 08fd829..0b27f52 100644
--- a/crates/daphne/src/roles/aggregator.rs
+++ b/crates/daphne/src/roles/aggregator.rs
@@ -15,7 +15,7 @@ use crate::{
     metrics::{DaphneMetrics, DaphneRequestType},
     protocol::aggregator::{EarlyReportStateConsumed, EarlyReportStateInitialized},
     taskprov, DapAggregateShare, DapAggregateSpan, DapAggregationParam, DapError, DapGlobalConfig,
-    DapRequest, DapResponse, DapTaskConfig, DapVersion,
+    DapRequestMeta, DapResponse, DapTaskConfig, DapVersion,
 };
 
 /// Report initializer. Used by a DAP Aggregator [`DapAggregator`] when initializing an aggregation
@@ -84,7 +84,7 @@ pub trait DapAggregator: HpkeProvider + DapReportInitializer + Sized {
     /// nothing.
     async fn taskprov_put(
         &self,
-        req: &DapRequest,
+        req: &DapRequestMeta,
         task_config: DapTaskConfig,
     ) -> Result<(), DapError>;
 
diff --git a/crates/daphne/src/roles/helper.rs b/crates/daphne/src/roles/helper.rs
index 0ab6f20..3d839e4 100644
--- a/crates/daphne/src/roles/helper.rs
+++ b/crates/daphne/src/roles/helper.rs
@@ -6,7 +6,7 @@ use std::{collections::HashMap, sync::Once};
 use async_trait::async_trait;
 use prio::codec::{Encode, ParameterizedDecode};
 
-use super::{check_batch, check_request_content_type, resolve_taskprov, DapAggregator};
+use super::{check_batch, resolve_taskprov, DapAggregator};
 use crate::{
     constants::DapMediaType,
     error::DapAbort,
@@ -25,23 +25,20 @@ use crate::{
 #[async_trait]
 pub trait DapHelper: DapAggregator {}
 
-pub async fn handle_agg_job_init_req<'req, A: DapHelper>(
+pub async fn handle_agg_job_init_req<A: DapHelper>(
     aggregator: &A,
-    req: &'req DapRequest,
+    req: DapRequest<AggregationJobInitReq>,
     replay_protection: ReplayProtection,
 ) -> Result<DapResponse, DapError> {
     let global_config = aggregator.get_global_config().await?;
     let task_id = req.task_id;
     let metrics = aggregator.metrics();
-    let agg_job_init_req =
-        AggregationJobInitReq::get_decoded_with_param(&req.version, &req.payload)
-            .map_err(|e| DapAbort::from_codec_error(e, task_id))?;
 
-    metrics.agg_job_observe_batch_size(agg_job_init_req.prep_inits.len());
+    metrics.agg_job_observe_batch_size(req.payload.prep_inits.len());
 
     // taskprov: Resolve the task config to use for the request.
     if global_config.allow_taskprov {
-        resolve_taskprov(aggregator, &task_id, req, &global_config).await?;
+        resolve_taskprov(aggregator, &task_id, &req, &global_config).await?;
     }
 
     let wrapped_task_config = aggregator
@@ -63,17 +60,18 @@ pub async fn handle_agg_job_init_req<'req, A: DapHelper>(
     check_part_batch(
         &task_id,
         task_config,
-        &agg_job_init_req.part_batch_sel,
-        &agg_job_init_req.agg_param,
+        &req.payload.part_batch_sel,
+        &req.payload.agg_param,
     )?;
 
-    let part_batch_sel = agg_job_init_req.part_batch_sel.clone();
+    let part_batch_sel = req.payload.part_batch_sel.clone();
+    let version = req.version;
     let initialized_reports = task_config
         .consume_agg_job_req(
             aggregator,
             aggregator,
             &task_id,
-            agg_job_init_req,
+            req.payload,
             replay_protection,
         )
         .await?;
@@ -103,41 +101,25 @@ pub async fn handle_agg_job_init_req<'req, A: DapHelper>(
 
     metrics.inbound_req_inc(DaphneRequestType::Aggregate);
     Ok(DapResponse {
-        version: req.version,
+        version,
         media_type: DapMediaType::AggregationJobResp,
         payload: agg_job_resp.get_encoded().map_err(DapError::encoding)?,
     })
 }
 
-/// Handle a request pertaining to an aggregation job.
-pub async fn handle_agg_job_req<'req, A: DapHelper>(
-    aggregator: &A,
-    req: &DapRequest,
-    replay_protection: ReplayProtection,
-) -> Result<DapResponse, DapError> {
-    match req.media_type {
-        Some(DapMediaType::AggregationJobInitReq) => {
-            handle_agg_job_init_req(aggregator, req, replay_protection).await
-        }
-        _ => Err(DapAbort::BadRequest("unexpected media type".into()).into()),
-    }
-}
-
 /// Handle a request for an aggregate share. This is called by the Leader to complete a
 /// collection job.
 pub async fn handle_agg_share_req<'req, A: DapHelper>(
     aggregator: &A,
-    req: &DapRequest,
+    req: DapRequest<AggregateShareReq>,
 ) -> Result<DapResponse, DapError> {
     let global_config = aggregator.get_global_config().await?;
     let now = aggregator.get_current_time();
     let metrics = aggregator.metrics();
     let task_id = req.task_id;
 
-    check_request_content_type(req, DapMediaType::AggregateShareReq)?;
-
     if global_config.allow_taskprov {
-        resolve_taskprov(aggregator, &task_id, req, &global_config).await?;
+        resolve_taskprov(aggregator, &task_id, &req, &global_config).await?;
     }
 
     let wrapped_task_config = aggregator
@@ -151,11 +133,8 @@ pub async fn handle_agg_share_req<'req, A: DapHelper>(
         return Err(DapAbort::version_mismatch(req.version, task_config.version).into());
     }
 
-    let agg_share_req = AggregateShareReq::get_decoded_with_param(&req.version, &req.payload)
-        .map_err(|e| DapAbort::from_codec_error(e, task_id))?;
-
     let agg_param =
-        DapAggregationParam::get_decoded_with_param(&task_config.vdaf, &agg_share_req.agg_param)
+        DapAggregationParam::get_decoded_with_param(&task_config.vdaf, &req.payload.agg_param)
             .map_err(|e| DapAbort::from_codec_error(e, task_id))?;
 
     // Ensure the batch boundaries are valid and that the batch doesn't overlap with previosuly
@@ -164,25 +143,25 @@ pub async fn handle_agg_share_req<'req, A: DapHelper>(
         aggregator,
         task_config,
         &task_id,
-        &agg_share_req.batch_sel.clone().into(),
-        &agg_share_req.agg_param,
+        &req.payload.batch_sel.clone().into(),
+        &req.payload.agg_param,
         now,
         &global_config,
     )
     .await?;
 
     let agg_share = aggregator
-        .get_agg_share(&task_id, &agg_share_req.batch_sel)
+        .get_agg_share(&task_id, &req.payload.batch_sel)
         .await?;
 
     // Check that we have aggreagted the same set of reports as the Leader.
-    if agg_share_req.report_count != agg_share.report_count
-        || !constant_time_eq(&agg_share_req.checksum, &agg_share.checksum)
+    if req.payload.report_count != agg_share.report_count
+        || !constant_time_eq(&req.payload.checksum, &agg_share.checksum)
     {
         return Err(DapAbort::BatchMismatch{
                 detail: format!("Either the report count or checksum does not match: the Leader computed {} and {}; the Helper computed {} and {}.",
-                    agg_share_req.report_count,
-                    hex::encode(agg_share_req.checksum),
+                    req.payload.report_count,
+                    hex::encode(req.payload.checksum),
                     agg_share.report_count,
                     hex::encode(agg_share.checksum)),
                 task_id,
@@ -206,13 +185,13 @@ pub async fn handle_agg_share_req<'req, A: DapHelper>(
 
     // Mark each aggregated report as collected.
     aggregator
-        .mark_collected(&task_id, &agg_share_req.batch_sel)
+        .mark_collected(&task_id, &req.payload.batch_sel)
         .await?;
 
     let encrypted_agg_share = task_config.produce_helper_encrypted_agg_share(
         &task_config.collector_hpke_config,
         &task_id,
-        &agg_share_req.batch_sel,
+        &req.payload.batch_sel,
         &agg_param,
         &agg_share,
         task_config.version,
@@ -222,7 +201,7 @@ pub async fn handle_agg_share_req<'req, A: DapHelper>(
         encrypted_agg_share,
     };
 
-    metrics.report_inc_by(ReportStatus::Collected, agg_share_req.report_count);
+    metrics.report_inc_by(ReportStatus::Collected, req.payload.report_count);
     metrics.inbound_req_inc(DaphneRequestType::Collect);
     Ok(DapResponse {
         version: req.version,
diff --git a/crates/daphne/src/roles/leader/mod.rs b/crates/daphne/src/roles/leader/mod.rs
index 31ceaec..0362cff 100644
--- a/crates/daphne/src/roles/leader/mod.rs
+++ b/crates/daphne/src/roles/leader/mod.rs
@@ -12,10 +12,7 @@ use rand::{thread_rng, Rng};
 use tracing::debug;
 use url::Url;
 
-use super::{
-    aggregator::MergeAggShareError, check_batch, check_request_content_type, resolve_taskprov,
-    DapAggregator,
-};
+use super::{aggregator::MergeAggShareError, check_batch, resolve_taskprov, DapAggregator};
 use crate::{
     constants::DapMediaType,
     error::DapAbort,
@@ -27,15 +24,15 @@ use crate::{
     },
     metrics::{DaphneRequestType, ReportStatus},
     DapAggregationParam, DapCollectionJob, DapError, DapLeaderProcessTelemetry, DapRequest,
-    DapResource, DapResponse, DapTaskConfig,
+    DapRequestMeta, DapResource, DapResponse, DapTaskConfig, DapVersion,
 };
 
-struct LeaderHttpRequestOptions<'p> {
+struct LeaderHttpRequestOptions<'p, P> {
     path: &'p str,
     req_media_type: DapMediaType,
     resp_media_type: DapMediaType,
     resource: DapResource,
-    req_data: Vec<u8>,
+    req_data: P,
     method: LeaderHttpRequestMethod,
     taskprov: Option<String>,
 }
@@ -45,12 +42,15 @@ enum LeaderHttpRequestMethod {
     Put,
 }
 
-async fn leader_send_http_request(
+async fn leader_send_http_request<P>(
     role: &impl DapLeader,
     task_id: &TaskId,
     task_config: &DapTaskConfig,
-    opts: LeaderHttpRequestOptions<'_>,
-) -> Result<DapResponse, DapError> {
+    opts: LeaderHttpRequestOptions<'_, P>,
+) -> Result<DapResponse, DapError>
+where
+    P: Send + ParameterizedEncode<DapVersion>,
+{
     let LeaderHttpRequestOptions {
         path,
         req_media_type,
@@ -67,12 +67,14 @@ async fn leader_send_http_request(
         .map_err(|e| fatal_error!(err = ?e, "failed to helper url {:?} with {path:?}", task_config.helper_url))?;
 
     let req = DapRequest {
-        version: task_config.version,
-        media_type: Some(req_media_type),
-        task_id: *task_id,
-        resource,
+        meta: DapRequestMeta {
+            version: task_config.version,
+            media_type: Some(req_media_type),
+            task_id: *task_id,
+            resource,
+            taskprov,
+        },
         payload: req_data,
-        taskprov,
     };
 
     let resp = match method {
@@ -154,30 +156,35 @@ pub trait DapLeader: DapAggregator {
     ) -> Result<(), DapError>;
 
     /// Send an HTTP POST request.
-    async fn send_http_post(&self, req: DapRequest, url: Url) -> Result<DapResponse, DapError>;
+    async fn send_http_post<M>(
+        &self,
+        req: DapRequest<M>,
+        url: Url,
+    ) -> Result<DapResponse, DapError>
+    where
+        M: Send + ParameterizedEncode<DapVersion>;
 
     /// Send an HTTP PUT request.
-    async fn send_http_put(&self, req: DapRequest, url: Url) -> Result<DapResponse, DapError>;
+    async fn send_http_put<M>(&self, req: DapRequest<M>, url: Url) -> Result<DapResponse, DapError>
+    where
+        M: Send + ParameterizedEncode<DapVersion>;
 }
 
 /// Handle a report from a Client.
 pub async fn handle_upload_req<A: DapLeader>(
     aggregator: &A,
-    req: &DapRequest,
+    req: DapRequest<Report>,
 ) -> Result<(), DapError> {
     let global_config = aggregator.get_global_config().await?;
     let metrics = aggregator.metrics();
     let task_id = req.task_id;
     debug!("upload for task {task_id}");
 
-    check_request_content_type(req, DapMediaType::Report)?;
-
-    let report = Report::get_decoded_with_param(&req.version, req.payload.as_ref())
-        .map_err(|e| DapAbort::from_codec_error(e, task_id))?;
+    let report = &req.payload;
     debug!("report id is {}", report.report_metadata.id);
 
     if global_config.allow_taskprov {
-        resolve_taskprov(aggregator, &task_id, req, &global_config).await?;
+        resolve_taskprov(aggregator, &task_id, &req, &global_config).await?;
     }
     let task_config = aggregator
         .get_task_config_for(&task_id)
@@ -230,7 +237,7 @@ pub async fn handle_upload_req<A: DapLeader>(
     // Store the report for future processing. At this point, the report may be rejected if
     // the Leader detects that the report was replayed or pertains to a batch that has already
     // been collected.
-    aggregator.put_report(&report, &task_id).await?;
+    aggregator.put_report(report, &task_id).await?;
 
     metrics.inbound_req_inc(DaphneRequestType::Upload);
     Ok(())
@@ -240,7 +247,7 @@ pub async fn handle_upload_req<A: DapLeader>(
 /// poll later on to get the collection.
 pub async fn handle_coll_job_req<A: DapLeader>(
     aggregator: &A,
-    req: &DapRequest,
+    req: &DapRequest<CollectionReq>,
 ) -> Result<Url, DapError> {
     let global_config = aggregator.get_global_config().await?;
     let now = aggregator.get_current_time();
@@ -248,8 +255,6 @@ pub async fn handle_coll_job_req<A: DapLeader>(
     let task_id = req.task_id;
     debug!("collect for task {task_id}");
 
-    check_request_content_type(req, DapMediaType::CollectReq)?;
-
     if global_config.allow_taskprov {
         resolve_taskprov(aggregator, &task_id, req, &global_config).await?;
     }
@@ -260,8 +265,7 @@ pub async fn handle_coll_job_req<A: DapLeader>(
         .ok_or(DapAbort::UnrecognizedTask { task_id })?;
     let task_config = wrapped_task_config.as_ref();
 
-    let coll_job_req = CollectionReq::get_decoded_with_param(&req.version, req.payload.as_ref())
-        .map_err(|e| DapAbort::from_codec_error(e, task_id))?;
+    let coll_job_req = &req.payload;
 
     let agg_param =
         DapAggregationParam::get_decoded_with_param(&task_config.vdaf, &coll_job_req.agg_param)
@@ -285,9 +289,7 @@ pub async fn handle_coll_job_req<A: DapLeader>(
     )
     .await?;
 
-    let DapResource::CollectionJob(coll_job_id) = &req.resource else {
-        return Err(DapAbort::BadRequest("missing collection ID".into()).into());
-    };
+    let coll_job_id = req.collection_job_id()?;
 
     let batch_sel = match coll_job_req.query {
         Query::TimeInterval { batch_interval } => BatchSelector::TimeInterval { batch_interval },
@@ -353,9 +355,7 @@ async fn run_agg_job<A: DapLeader>(
             req_media_type: DapMediaType::AggregationJobInitReq,
             resp_media_type: DapMediaType::AggregationJobResp,
             resource: DapResource::AggregationJob(agg_job_id),
-            req_data: agg_job_init_req
-                .get_encoded_with_param(&task_config.version)
-                .map_err(DapError::encoding)?,
+            req_data: agg_job_init_req,
             method: LeaderHttpRequestMethod::Put,
             taskprov: taskprov.clone(),
         },
@@ -457,6 +457,8 @@ async fn run_coll_job<A: DapLeader>(
 
     let url_path = format!("tasks/{}/aggregate_shares", task_id.to_base64url());
 
+    let agg_share_req_report_count = agg_share_req.report_count;
+    let agg_share_req_batch_sel = agg_share_req.batch_sel.clone();
     // Send AggregateShareReq and receive AggregateShareResp.
     let resp = leader_send_http_request(
         aggregator,
@@ -467,9 +469,7 @@ async fn run_coll_job<A: DapLeader>(
             req_media_type: DapMediaType::AggregateShareReq,
             resp_media_type: DapMediaType::AggregateShare,
             resource: DapResource::Undefined,
-            req_data: agg_share_req
-                .get_encoded_with_param(&task_config.version)
-                .map_err(DapError::encoding)?,
+            req_data: agg_share_req,
             method: LeaderHttpRequestMethod::Post,
             taskprov,
         },
@@ -506,11 +506,11 @@ async fn run_coll_job<A: DapLeader>(
 
     // Mark reports as collected.
     aggregator
-        .mark_collected(task_id, &agg_share_req.batch_sel)
+        .mark_collected(task_id, &agg_share_req_batch_sel)
         .await?;
 
-    metrics.report_inc_by(ReportStatus::Collected, agg_share_req.report_count);
-    Ok(agg_share_req.report_count)
+    metrics.report_inc_by(ReportStatus::Collected, agg_share_req_report_count);
+    Ok(agg_share_req_report_count)
 }
 
 /// Drain a number of items from the work queue and process them.
diff --git a/crates/daphne/src/roles/mod.rs b/crates/daphne/src/roles/mod.rs
index 85d0577..6a6a638 100644
--- a/crates/daphne/src/roles/mod.rs
+++ b/crates/daphne/src/roles/mod.rs
@@ -8,9 +8,8 @@ pub mod helper;
 pub mod leader;
 
 use crate::{
-    constants::DapMediaType,
     messages::{Base64Encode, Query, TaskId, Time},
-    taskprov, DapAbort, DapError, DapGlobalConfig, DapQueryConfig, DapRequest, DapTaskConfig,
+    taskprov, DapAbort, DapError, DapGlobalConfig, DapQueryConfig, DapRequestMeta, DapTaskConfig,
 };
 use tracing::warn;
 
@@ -92,18 +91,10 @@ async fn check_batch(
     Ok(())
 }
 
-fn check_request_content_type(req: &DapRequest, expected: DapMediaType) -> Result<(), DapAbort> {
-    if req.media_type != Some(expected) {
-        Err(DapAbort::content_type(req, expected))
-    } else {
-        Ok(())
-    }
-}
-
 async fn resolve_taskprov(
     agg: &impl DapAggregator,
     task_id: &TaskId,
-    req: &DapRequest,
+    req: &DapRequestMeta,
     global_config: &DapGlobalConfig,
 ) -> Result<(), DapError> {
     if agg.get_task_config_for(task_id).await?.is_some() {
@@ -160,12 +151,12 @@ mod test {
         testing::InMemoryAggregator,
         vdaf::{Prio3Config, VdafConfig},
         DapAbort, DapAggregationJobState, DapAggregationParam, DapBatchBucket, DapCollectionJob,
-        DapError, DapGlobalConfig, DapMeasurement, DapQueryConfig, DapRequest, DapResource,
-        DapTaskConfig, DapTaskParameters, DapVersion,
+        DapError, DapGlobalConfig, DapMeasurement, DapQueryConfig, DapRequest, DapRequestMeta,
+        DapResource, DapTaskConfig, DapTaskParameters, DapVersion,
     };
     use assert_matches::assert_matches;
     use matchit::Router;
-    use prio::codec::{Decode, Encode, ParameterizedEncode};
+    use prio::codec::{Decode, Encode};
     #[cfg(feature = "experimental")]
     use prio::{idpf::IdpfInput, vdaf::poplar1::Poplar1AggregationParam};
     use rand::{thread_rng, Rng};
@@ -402,21 +393,31 @@ mod test {
             data.with_leader(helper)
         }
 
-        pub async fn gen_test_upload_req(&self, report: Report, task_id: &TaskId) -> DapRequest {
+        pub async fn gen_test_upload_req(
+            &self,
+            report: Report,
+            task_id: &TaskId,
+        ) -> DapRequest<Report> {
             let task_config = self.leader.unchecked_get_task_config(task_id).await;
             let version = task_config.version;
 
             DapRequest {
-                version,
-                media_type: Some(DapMediaType::Report),
-                task_id: *task_id,
-                resource: DapResource::Undefined,
-                payload: report.get_encoded_with_param(&version).unwrap(),
-                ..Default::default()
+                meta: DapRequestMeta {
+                    version,
+                    media_type: Some(DapMediaType::Report),
+                    task_id: *task_id,
+                    resource: DapResource::Undefined,
+                    ..Default::default()
+                },
+                payload: report,
             }
         }
 
-        pub async fn gen_test_coll_job_req(&self, query: Query, task_id: &TaskId) -> DapRequest {
+        pub async fn gen_test_coll_job_req(
+            &self,
+            query: Query,
+            task_id: &TaskId,
+        ) -> DapRequest<CollectionReq> {
             self.gen_test_coll_job_req_for_collection(query, DapAggregationParam::Empty, task_id)
                 .await
         }
@@ -426,13 +427,13 @@ mod test {
             query: Query,
             agg_param: DapAggregationParam,
             task_id: &TaskId,
-        ) -> DapRequest {
+        ) -> DapRequest<CollectionReq> {
             let task_config = self.leader.unchecked_get_task_config(task_id).await;
 
             Self::collector_req(
                 task_id,
                 &task_config,
-                DapMediaType::CollectReq,
+                DapMediaType::CollectionReq,
                 CollectionReq {
                     query,
                     agg_param: agg_param.get_encoded().unwrap(),
@@ -445,7 +446,7 @@ mod test {
             task_id: &TaskId,
             agg_param: DapAggregationParam,
             reports: Vec<Report>,
-        ) -> (DapAggregationJobState, DapRequest) {
+        ) -> (DapAggregationJobState, DapRequest<AggregationJobInitReq>) {
             let mut rng = thread_rng();
             let task_config = self.leader.unchecked_get_task_config(task_id).await;
             let part_batch_sel = match task_config.query {
@@ -525,42 +526,45 @@ mod test {
                 .unwrap()
         }
 
-        pub fn leader_req<M: ParameterizedEncode<DapVersion>>(
+        pub fn leader_req<P>(
             task_id: &TaskId,
             task_config: &DapTaskConfig,
             agg_job_id: Option<&AggregationJobId>,
             media_type: DapMediaType,
-            msg: M,
-        ) -> DapRequest {
-            let payload = msg.get_encoded_with_param(&task_config.version).unwrap();
+            payload: P,
+        ) -> DapRequest<P> {
             DapRequest {
-                version: task_config.version,
-                media_type: Some(media_type),
-                task_id: *task_id,
-                resource: agg_job_id.map_or(DapResource::Undefined, |id| {
-                    DapResource::AggregationJob(*id)
-                }),
+                meta: DapRequestMeta {
+                    version: task_config.version,
+                    media_type: Some(media_type),
+                    task_id: *task_id,
+                    resource: agg_job_id.map_or(DapResource::Undefined, |id| {
+                        DapResource::AggregationJob(*id)
+                    }),
+                    ..Default::default()
+                },
                 payload,
-                ..Default::default()
             }
         }
 
-        pub fn collector_req<M: ParameterizedEncode<DapVersion>>(
+        pub fn collector_req<C>(
             task_id: &TaskId,
             task_config: &DapTaskConfig,
             media_type: DapMediaType,
-            msg: M,
-        ) -> DapRequest {
+            payload: C,
+        ) -> DapRequest<C> {
             let mut rng = thread_rng();
             let coll_job_id = CollectionJobId(rng.gen());
 
             DapRequest {
-                version: task_config.version,
-                media_type: Some(media_type),
-                task_id: *task_id,
-                resource: DapResource::CollectionJob(coll_job_id),
-                payload: msg.get_encoded_with_param(&task_config.version).unwrap(),
-                ..Default::default()
+                meta: DapRequestMeta {
+                    version: task_config.version,
+                    media_type: Some(media_type),
+                    task_id: *task_id,
+                    resource: DapResource::CollectionJob(coll_job_id),
+                    ..Default::default()
+                },
+                payload,
             }
         }
     }
@@ -588,7 +592,7 @@ mod test {
             },
         );
         assert_matches!(
-            helper::handle_agg_job_req(&*t.helper, &req, Default::default())
+            helper::handle_agg_job_init_req(&*t.helper, req, Default::default())
                 .await
                 .unwrap_err(),
             DapError::Abort(DapAbort::QueryMismatch { .. })
@@ -633,17 +637,9 @@ mod test {
         let t = Test::new(version);
         let mut rng = thread_rng();
         let task_id = TaskId(rng.gen());
-        let req = DapRequest {
-            version,
-            media_type: Some(DapMediaType::HpkeConfigList),
-            payload: Vec::new(),
-            task_id,
-            resource: DapResource::Undefined,
-            ..Default::default()
-        };
 
         assert_eq!(
-            aggregator::handle_hpke_config_req(&*t.leader, req.version, Some(task_id))
+            aggregator::handle_hpke_config_req(&*t.leader, version, Some(task_id))
                 .await
                 .unwrap_err(),
             DapError::Abort(DapAbort::UnrecognizedTask { task_id })
@@ -654,20 +650,12 @@ mod test {
 
     async fn handle_hpke_config_req_missing_task_id(version: DapVersion) {
         let t = Test::new(version);
-        let req = DapRequest {
-            version,
-            media_type: Some(DapMediaType::HpkeConfigList),
-            task_id: t.time_interval_task_id,
-            resource: DapResource::Undefined,
-            payload: Vec::new(),
-            ..Default::default()
-        };
 
         // An Aggregator is permitted to abort an HPKE config request if the task ID is missing. Note
         // that Daphne-Workder does not implement this behavior. Instead it returns the HPKE config
         // used for all tasks.
         assert_matches!(
-            aggregator::handle_hpke_config_req(&*t.leader, req.version, None).await,
+            aggregator::handle_hpke_config_req(&*t.leader, version, None).await,
             Err(DapError::Abort(DapAbort::MissingTaskId))
         );
     }
@@ -699,7 +687,7 @@ mod test {
             },
         );
         assert_matches!(
-            helper::handle_agg_share_req(&*t.helper, &req)
+            helper::handle_agg_share_req(&*t.helper, req)
                 .await
                 .unwrap_err(),
             DapError::Abort(DapAbort::QueryMismatch { .. })
@@ -725,7 +713,7 @@ mod test {
             },
         );
         assert_matches!(
-            helper::handle_agg_share_req(&*t.helper, &req)
+            helper::handle_agg_share_req(&*t.helper, req)
                 .await
                 .unwrap_err(),
             DapError::Abort(DapAbort::BatchInvalid { .. })
@@ -746,7 +734,7 @@ mod test {
 
         // Get AggregationJobResp and then extract the transition data from inside.
         let agg_job_resp = AggregationJobResp::get_decoded(
-            &helper::handle_agg_job_req(&*t.helper, &req, Default::default())
+            &helper::handle_agg_job_init_req(&*t.helper, req, Default::default())
                 .await
                 .unwrap()
                 .payload,
@@ -774,7 +762,7 @@ mod test {
 
         // Get AggregationJobResp and then extract the transition data from inside.
         let agg_job_resp = AggregationJobResp::get_decoded(
-            &helper::handle_agg_job_req(&*t.helper, &req, Default::default())
+            &helper::handle_agg_job_init_req(&*t.helper, req, Default::default())
                 .await
                 .unwrap()
                 .payload,
@@ -795,7 +783,7 @@ mod test {
 
         let report = t.gen_test_report(task_id).await;
         let req = t.gen_test_upload_req(report.clone(), task_id).await;
-        leader::handle_upload_req(&*t.leader, &req).await.unwrap();
+        leader::handle_upload_req(&*t.leader, req).await.unwrap();
 
         let query = task_config.query_for_current_batch_window(t.now);
         let req = t.gen_test_coll_job_req(query, task_id).await;
@@ -842,7 +830,7 @@ mod test {
 
         let report = t.gen_test_report(task_id).await;
         let req = t.gen_test_upload_req(report.clone(), task_id).await;
-        leader::handle_upload_req(&*t.leader, &req).await.unwrap();
+        leader::handle_upload_req(&*t.leader, req).await.unwrap();
 
         // Add mock data to the aggreagte store backend. This is done in its own scope so that the lock
         // is released before running the test. Otherwise the test will deadlock.
@@ -887,19 +875,19 @@ mod test {
         // Construct a report payload with an invalid task ID.
         let report_invalid_task_id = t.gen_test_report(task_id).await;
         let req = DapRequest {
-            version: task_config.version,
-            media_type: Some(DapMediaType::Report),
-            task_id: TaskId([0; 32]),
-            resource: DapResource::Undefined,
-            payload: report_invalid_task_id
-                .get_encoded_with_param(&task_config.version)
-                .unwrap(),
-            ..Default::default()
+            meta: DapRequestMeta {
+                version: task_config.version,
+                media_type: Some(DapMediaType::Report),
+                task_id: TaskId([0; 32]),
+                resource: DapResource::Undefined,
+                ..Default::default()
+            },
+            payload: report_invalid_task_id,
         };
 
         // Expect failure due to invalid task ID in report.
         assert_eq!(
-            leader::handle_upload_req(&*t.leader, &req).await,
+            leader::handle_upload_req(&*t.leader, req).await,
             Err(DapError::Abort(DapAbort::UnrecognizedTask {
                 task_id: TaskId([0; 32])
             }))
@@ -916,16 +904,18 @@ mod test {
 
         let report = t.gen_test_report(task_id).await;
         let req = DapRequest {
-            version: task_config.version,
-            media_type: Some(DapMediaType::Report),
-            task_id: *task_id,
-            resource: DapResource::Undefined,
-            payload: report.get_encoded_with_param(&version).unwrap(),
-            ..Default::default()
+            meta: DapRequestMeta {
+                version: task_config.version,
+                media_type: Some(DapMediaType::Report),
+                task_id: *task_id,
+                resource: DapResource::Undefined,
+                ..Default::default()
+            },
+            payload: report.clone(),
         };
 
         assert_eq!(
-            leader::handle_upload_req(&*t.leader, &req)
+            leader::handle_upload_req(&*t.leader, req)
                 .await
                 .unwrap_err(),
             DapError::Abort(DapAbort::ReportTooLate {
@@ -944,7 +934,7 @@ mod test {
         for _ in 0..10 {
             let report = t.gen_test_report(task_id).await;
             let req = t.gen_test_upload_req(report.clone(), task_id).await;
-            leader::handle_upload_req(&*t.leader, &req).await.unwrap();
+            leader::handle_upload_req(&*t.leader, req).await.unwrap();
         }
 
         let query = task_config.query_for_current_batch_window(t.now);
@@ -997,7 +987,7 @@ mod test {
         let req = Test::collector_req(
             task_id,
             &task_config,
-            DapMediaType::CollectReq,
+            DapMediaType::CollectionReq,
             CollectionReq {
                 query: task_config.query_for_current_batch_window(t.now),
                 agg_param: Vec::default(),
@@ -1083,7 +1073,7 @@ mod test {
         let req = Test::collector_req(
             task_id,
             &task_config,
-            DapMediaType::CollectReq,
+            DapMediaType::CollectionReq,
             CollectionReq {
                 query: Query::TimeInterval {
                     batch_interval: Interval {
@@ -1108,7 +1098,7 @@ mod test {
         let req = Test::collector_req(
             task_id,
             &task_config,
-            DapMediaType::CollectReq,
+            DapMediaType::CollectionReq,
             CollectionReq {
                 query: Query::TimeInterval {
                     batch_interval: Interval {
@@ -1134,7 +1124,7 @@ mod test {
         let req = Test::collector_req(
             task_id,
             &task_config,
-            DapMediaType::CollectReq,
+            DapMediaType::CollectionReq,
             CollectionReq {
                 query: Query::TimeInterval {
                     batch_interval: Interval {
@@ -1168,7 +1158,7 @@ mod test {
         let req = Test::collector_req(
             task_id,
             &task_config,
-            DapMediaType::CollectReq,
+            DapMediaType::CollectionReq,
             CollectionReq {
                 query: Query::TimeInterval {
                     batch_interval: Interval {
@@ -1194,7 +1184,7 @@ mod test {
 
         let report = t.gen_test_report(task_id).await;
         let req = t.gen_test_upload_req(report.clone(), task_id).await;
-        leader::handle_upload_req(&*t.leader, &req).await.unwrap();
+        leader::handle_upload_req(&*t.leader, req).await.unwrap();
 
         let query = task_config.query_for_current_batch_window(t.now);
         let req = t.gen_test_coll_job_req(query, task_id).await;
@@ -1254,7 +1244,7 @@ mod test {
         let req = Test::collector_req(
             task_id,
             &task_config,
-            DapMediaType::CollectReq,
+            DapMediaType::CollectionReq,
             collector_collect_req.clone(),
         );
 
@@ -1307,7 +1297,7 @@ mod test {
         let req = Test::collector_req(
             &t.time_interval_task_id,
             &task_config,
-            DapMediaType::CollectReq,
+            DapMediaType::CollectionReq,
             CollectionReq {
                 query: Query::FixedSizeByBatchId {
                     batch_id: BatchId(rng.gen()),
@@ -1330,7 +1320,7 @@ mod test {
         let req = Test::collector_req(
             &t.fixed_size_task_id,
             &task_config,
-            DapMediaType::CollectReq,
+            DapMediaType::CollectionReq,
             CollectionReq {
                 query: Query::FixedSizeByBatchId {
                     batch_id: BatchId(rng.gen()), // Unrecognized batch ID
@@ -1355,7 +1345,7 @@ mod test {
         let report = t.gen_test_report(task_id).await;
         let req = t.gen_test_upload_req(report, task_id).await;
 
-        leader::handle_upload_req(&*t.leader, &req)
+        leader::handle_upload_req(&*t.leader, req)
             .await
             .expect("upload failed unexpectedly");
     }
@@ -1372,7 +1362,7 @@ mod test {
 
         // Client: Send upload request to Leader.
         let report = t.gen_test_report(task_id).await;
-        leader::handle_upload_req(&*t.leader, &t.gen_test_upload_req(report, task_id).await)
+        leader::handle_upload_req(&*t.leader, t.gen_test_upload_req(report, task_id).await)
             .await
             .unwrap();
 
@@ -1408,7 +1398,7 @@ mod test {
 
         // Client: Send upload request to Leader.
         let report = t.gen_test_report(task_id).await;
-        leader::handle_upload_req(&*t.leader, &t.gen_test_upload_req(report, task_id).await)
+        leader::handle_upload_req(&*t.leader, t.gen_test_upload_req(report, task_id).await)
             .await
             .unwrap();
 
@@ -1489,14 +1479,16 @@ mod test {
                 .unwrap();
 
             let req = DapRequest {
-                version,
-                media_type: Some(DapMediaType::Report),
-                task_id,
-                resource: DapResource::Undefined,
-                payload: report.get_encoded_with_param(&version).unwrap(),
-                taskprov: Some(taskprov_advertisement.clone()),
+                meta: DapRequestMeta {
+                    version,
+                    media_type: Some(DapMediaType::Report),
+                    task_id,
+                    resource: DapResource::Undefined,
+                    taskprov: Some(taskprov_advertisement.clone()),
+                },
+                payload: report,
             };
-            leader::handle_upload_req(&*t.leader, &req).await.unwrap();
+            leader::handle_upload_req(&*t.leader, req).await.unwrap();
         }
 
         // Collector: Request result from the Leader.
@@ -1611,7 +1603,7 @@ mod test {
                 .await;
 
             let report = t.gen_test_report(task_id).await;
-            leader::handle_upload_req(&*t.leader, &t.gen_test_upload_req(report, task_id).await)
+            leader::handle_upload_req(&*t.leader, t.gen_test_upload_req(report, task_id).await)
                 .await
                 .unwrap();
 
@@ -1626,7 +1618,7 @@ mod test {
             let task_id = &t.fixed_size_task_id;
 
             let report = t.gen_test_report(task_id).await;
-            leader::handle_upload_req(&*t.leader, &t.gen_test_upload_req(report, task_id).await)
+            leader::handle_upload_req(&*t.leader, t.gen_test_upload_req(report, task_id).await)
                 .await
                 .unwrap();
 
@@ -1677,7 +1669,7 @@ mod test {
                     },
                 )
                 .await;
-            leader::handle_upload_req(&*t.leader, &t.gen_test_upload_req(report, task_id).await)
+            leader::handle_upload_req(&*t.leader, t.gen_test_upload_req(report, task_id).await)
                 .await
                 .unwrap();
         }
diff --git a/crates/daphne/src/taskprov.rs b/crates/daphne/src/taskprov.rs
index c7a5508..ea58098 100644
--- a/crates/daphne/src/taskprov.rs
+++ b/crates/daphne/src/taskprov.rs
@@ -8,6 +8,7 @@
 use std::num::NonZeroUsize;
 
 use crate::{
+    error::DapAbort,
     fatal_error,
     hpke::HpkeConfig,
     messages::{
@@ -16,12 +17,15 @@ use crate::{
         Duration, TaskId, Time,
     },
     vdaf::VdafVerifyKey,
-    DapAbort, DapError, DapQueryConfig, DapRequest, DapTaskConfig, DapTaskConfigMethod, DapVersion,
-    Prio3Config, VdafConfig,
+    DapError, DapQueryConfig, DapRequestMeta, DapTaskConfig, DapTaskConfigMethod, DapVersion,
+    VdafConfig,
 };
 use crate::{
     pine::PineParam,
-    vdaf::pine::{pine32_hmac_sha256_aes128, pine64_hmac_sha256_aes128, PineConfig},
+    vdaf::{
+        pine::{pine32_hmac_sha256_aes128, pine64_hmac_sha256_aes128, PineConfig},
+        Prio3Config,
+    },
 };
 use prio::codec::ParameterizedDecode;
 use ring::{
@@ -107,7 +111,7 @@ fn malformed_task_config(task_id: &TaskId, detail: String) -> DapAbort {
 /// The `task_id` is the task ID indicated by the request; if this does not match the derived task
 /// ID, then we return `Err(DapError::Abort(DapAbort::UnrecognizedTask))`.
 pub(crate) fn resolve_advertised_task_config(
-    req: &'_ DapRequest,
+    req: &DapRequestMeta,
     verify_key_init: &[u8; 32],
     collector_hpke_config: &HpkeConfig,
     task_id: &TaskId,
@@ -127,7 +131,7 @@ pub(crate) fn resolve_advertised_task_config(
 
 /// Check for a taskprov extension in the report, and return it if found.
 fn get_taskprov_task_config(
-    req: &'_ DapRequest,
+    req: &DapRequestMeta,
     task_id: &TaskId,
 ) -> Result<Option<TaskConfig>, DapAbort> {
     let taskprov_data = if let Some(ref taskprov_base64url) = req.taskprov {
@@ -509,7 +513,7 @@ mod test {
         taskprov::{DapTaskConfigNeedsOptIn, OptInParam},
         test_versions,
         vdaf::{VdafConfig, VdafVerifyKey},
-        DapRequest, DapResource, DapVersion,
+        DapRequestMeta, DapResource, DapVersion,
     };
 
     /// Test conversion between the serialized task configuration and a `DapTaskConfig`.
@@ -621,13 +625,12 @@ mod test {
             let task_id = compute_task_id(&taskprov_task_config_bytes);
             let taskprov_task_config_base64url = encode_base64url(&taskprov_task_config_bytes);
 
-            let req = DapRequest {
+            let req = DapRequestMeta {
                 version,
-                media_type: None, // ignored by test
                 task_id,
-                resource: DapResource::Undefined, // ignored by test
-                payload: Vec::default(),          // ignored by test
                 taskprov: Some(taskprov_task_config_base64url),
+                media_type: None,                 // ignored by test
+                resource: DapResource::Undefined, // ignored by test
             };
 
             (req, task_id)
@@ -676,13 +679,12 @@ mod test {
             let task_id = compute_task_id(&taskprov_task_config_bytes);
             let taskprov_task_config_base64url = encode_base64url(&taskprov_task_config_bytes);
 
-            let req = DapRequest {
+            let req = crate::DapRequestMeta {
                 version,
-                media_type: None, // ignored by test
                 task_id,
-                resource: DapResource::Undefined, // ignored by test
-                payload: Vec::default(),          // ignored by test
                 taskprov: Some(taskprov_task_config_base64url),
+                media_type: None,
+                resource: DapResource::Undefined,
             };
 
             (req, task_id)


Test patch:
diff --git a/crates/daphne-server/tests/e2e/e2e.rs b/crates/daphne-server/tests/e2e/e2e.rs
index 6198cc1..39df895 100644
--- a/crates/daphne-server/tests/e2e/e2e.rs
+++ b/crates/daphne-server/tests/e2e/e2e.rs
@@ -788,7 +788,7 @@ async fn leader_collect_abort_invalid_batch_interval(version: DapVersion) {
         client,
         Some(&t.collector_bearer_token),
         path,
-        DapMediaType::CollectReq,
+        DapMediaType::CollectionReq,
         collect_req.get_encoded_with_param(&t.version).unwrap(),
         400,
         "batchInvalid",
@@ -810,7 +810,7 @@ async fn leader_collect_abort_invalid_batch_interval(version: DapVersion) {
         client,
         Some(&t.collector_bearer_token),
         path,
-        DapMediaType::CollectReq,
+        DapMediaType::CollectionReq,
         collect_req.get_encoded_with_param(&t.version).unwrap(),
         400,
         "batchInvalid",
@@ -900,7 +900,7 @@ async fn leader_collect_abort_overlapping_batch_interval(version: DapVersion) {
         client,
         Some(&t.collector_bearer_token),
         path,
-        DapMediaType::CollectReq,
+        DapMediaType::CollectionReq,
         collect_req.get_encoded_with_param(&t.version).unwrap(),
         400,
         "batchOverlap",
@@ -1070,7 +1070,7 @@ async fn fixed_size() {
         client,
         Some(&t.collector_bearer_token),
         &TestRunner::collect_path_for_task(&t.task_id),
-        DapMediaType::CollectReq,
+        DapMediaType::CollectionReq,
         CollectionReq {
             query: Query::FixedSizeByBatchId {
                 batch_id: prev_batch_id,
diff --git a/crates/daphne-server/tests/e2e/test_runner.rs b/crates/daphne-server/tests/e2e/test_runner.rs
index a1e6ef4..cc27207 100644
--- a/crates/daphne-server/tests/e2e/test_runner.rs
+++ b/crates/daphne-server/tests/e2e/test_runner.rs
@@ -590,7 +590,7 @@ impl TestRunner {
         headers.insert(
             reqwest::header::CONTENT_TYPE,
             reqwest::header::HeaderValue::from_str(
-                DapMediaType::CollectReq
+                DapMediaType::CollectionReq
                     .as_str_for_version(self.version)
                     .context("no string for version")?,
             )?,
@@ -779,7 +779,7 @@ impl TestRunner {
         headers.insert(
             reqwest::header::CONTENT_TYPE,
             reqwest::header::HeaderValue::from_str(
-                DapMediaType::CollectReq
+                DapMediaType::CollectionReq
                     .as_str_for_version(self.version)
                     .context("no string for version")?,
             )?,
diff --git a/crates/daphne/src/testing/mod.rs b/crates/daphne/src/testing/mod.rs
index a955ea2..f2f4ff4 100644
--- a/crates/daphne/src/testing/mod.rs
+++ b/crates/daphne/src/testing/mod.rs
@@ -28,11 +28,12 @@ use crate::{
     vdaf::VdafVerifyKey,
     DapAbort, DapAggregateResult, DapAggregateShare, DapAggregateSpan, DapAggregationJobState,
     DapAggregationParam, DapBatchBucket, DapCollectionJob, DapError, DapGlobalConfig,
-    DapMeasurement, DapQueryConfig, DapRequest, DapResponse, DapTaskConfig, DapVersion,
-    ReplayProtection, VdafConfig,
+    DapMeasurement, DapQueryConfig, DapRequest, DapRequestMeta, DapResponse, DapTaskConfig,
+    DapVersion, ReplayProtection, VdafConfig,
 };
 use async_trait::async_trait;
 use deepsize::DeepSizeOf;
+use prio::codec::{ParameterizedDecode, ParameterizedEncode};
 use rand::{thread_rng, Rng};
 use serde::{Deserialize, Serialize};
 use std::{
@@ -798,7 +799,7 @@ impl DapAggregator for InMemoryAggregator {
 
     async fn taskprov_put(
         &self,
-        req: &DapRequest,
+        req: &DapRequestMeta,
         task_config: DapTaskConfig,
     ) -> Result<(), DapError> {
         let mut tasks = self.tasks.lock().expect("tasks: lock failed");
@@ -1067,18 +1068,25 @@ impl DapLeader for InMemoryAggregator {
             .finish_collect_job(task_id, coll_job_id, collection)
     }
 
-    async fn send_http_post(&self, req: DapRequest, _url: Url) -> Result<DapResponse, DapError> {
+    async fn send_http_post<M>(
+        &self,
+        req: DapRequest<M>,
+        _url: Url,
+    ) -> Result<DapResponse, DapError>
+    where
+        M: Send + ParameterizedEncode<DapVersion>,
+    {
         match req.media_type {
-            Some(DapMediaType::AggregationJobInitReq) => Ok(helper::handle_agg_job_req(
+            Some(DapMediaType::AggregationJobInitReq) => Ok(helper::handle_agg_job_init_req(
                 &**self.peer.as_ref().expect("peer not configured"),
-                &req,
+                map_dap_req(req),
                 Default::default(),
             )
             .await
             .expect("peer aborted unexpectedly")),
             Some(DapMediaType::AggregateShareReq) => Ok(helper::handle_agg_share_req(
                 &**self.peer.as_ref().expect("peer not configured"),
-                &req,
+                map_dap_req(req),
             )
             .await
             .expect("peer aborted unexpectedly")),
@@ -1086,11 +1094,14 @@ impl DapLeader for InMemoryAggregator {
         }
     }
 
-    async fn send_http_put(&self, req: DapRequest, _url: Url) -> Result<DapResponse, DapError> {
+    async fn send_http_put<P>(&self, req: DapRequest<P>, _url: Url) -> Result<DapResponse, DapError>
+    where
+        P: Send + ParameterizedEncode<DapVersion>,
+    {
         if req.media_type == Some(DapMediaType::AggregationJobInitReq) {
-            Ok(helper::handle_agg_job_req(
+            Ok(helper::handle_agg_job_init_req(
                 &**self.peer.as_ref().expect("peer not configured"),
-                &req,
+                map_dap_req(req),
                 Default::default(),
             )
             .await
@@ -1101,6 +1112,24 @@ impl DapLeader for InMemoryAggregator {
     }
 }
 
+// this is effectively simulates the network boundary the request goes through. We must write to
+// the wire (ParameterizedEncode) and decode from the wire (ParameterizedDecode).
+//
+// This is only correct if I is the same type as O. But this can't be checked at compile time.
+fn map_dap_req<I: ParameterizedEncode<DapVersion>, O: ParameterizedDecode<DapVersion>>(
+    req: DapRequest<I>,
+) -> DapRequest<O> {
+    let version = req.version;
+    DapRequest {
+        meta: req.meta,
+        payload: O::get_decoded_with_param(
+            &version,
+            &req.payload.get_encoded_with_param(&version).unwrap(),
+        )
+        .unwrap(),
+    }
+}
+
 /// Information associated to a certain helper state for a given task ID and aggregate job ID.
 #[derive(Clone, Eq, Hash, PartialEq, Deserialize, Serialize)]
 #[cfg_attr(any(test, feature = "test-utils"), derive(deepsize::DeepSizeOf))]