Repository: cloudflare/daphne
Pull request #690 (2024-10-02 16:48:39)

In server examples output the config as yaml instead of rust debug


Patch:
diff --git a/.github/workflows/daphneci.yml b/.github/workflows/daphneci.yml
index 81da590..d0f7f7c 100644
--- a/.github/workflows/daphneci.yml
+++ b/.github/workflows/daphneci.yml
@@ -32,8 +32,6 @@ jobs:
         run: cargo fmt --all --check
       - name: Documentation
         run: RUSTDOCFLAGS='-D warnings' cargo doc --locked --no-deps --all-features --workspace
-      - name: Build
-        run: cargo build --all-targets
       - name: Linting
         run: cargo hack clippy --locked --each-feature -- -D warnings
       - name: Linting Tests
diff --git a/Cargo.lock b/Cargo.lock
index 9f61bad..b30fe73 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -955,6 +955,7 @@ dependencies = [
  "daphne",
  "daphne-service-utils",
  "dhat",
+ "either",
  "futures",
  "hex",
  "hpke-rs",
@@ -971,6 +972,7 @@ dependencies = [
  "reqwest 0.11.27",
  "serde",
  "serde_json",
+ "serde_yaml",
  "thiserror",
  "tokio",
  "tower",
@@ -3391,6 +3393,19 @@ dependencies = [
  "serde",
 ]
 
+[[package]]
+name = "serde_yaml"
+version = "0.9.34+deprecated"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47"
+dependencies = [
+ "indexmap",
+ "itoa",
+ "ryu",
+ "serde",
+ "unsafe-libyaml",
+]
+
 [[package]]
 name = "sha1"
 version = "0.10.6"
@@ -3983,6 +3998,12 @@ dependencies = [
  "subtle",
 ]
 
+[[package]]
+name = "unsafe-libyaml"
+version = "0.2.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861"
+
 [[package]]
 name = "untrusted"
 version = "0.9.0"
diff --git a/Cargo.toml b/Cargo.toml
index 0c715bb..8633bbc 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -46,6 +46,7 @@ constcat = "0.5.0"
 criterion = { version = "0.5.1", features = ["async_tokio"] }
 deepsize = { version = "0.2.0" }
 dhat = "0.3.3"
+either = "1.13.0"
 futures = "0.3.30"
 getrandom = "0.2.15"
 headers = "0.4"
@@ -77,6 +78,7 @@ rustls-pemfile = "2.1.3"
 serde = { version = "1.0.203", features = ["derive"] }
 serde-wasm-bindgen = "0.6.5"
 serde_json = "1.0.118"
+serde_yaml = "0.9.33"
 strum = { version = "0.26.3", features = ["derive"] }
 thiserror = "1.0.61"
 tokio = { version = "1.38.0", features = ["macros", "rt-multi-thread"] }
@@ -86,9 +88,9 @@ tracing = "0.1.40"
 tracing-core = "0.1.32"
 tracing-subscriber = "0.3.18"
 url = { version = "2.5.2", features = ["serde"] }
+wasm-streams = "0.4"
 webpki = "0.22.4"
 worker = { version = "0.3.3", features = ["http"] }
-wasm-streams = "0.4"
 x509-parser = "0.15.1"
 
 [workspace.dependencies.sentry]
diff --git a/Makefile b/Makefile
index 8516dfc..8688938 100644
--- a/Makefile
+++ b/Makefile
@@ -18,7 +18,12 @@ s: storage-proxy
 e2e: /tmp/private-key /tmp/certificate
 	export HPKE_SIGNING_KEY="$$(cat /tmp/private-key)"; \
 	export E2E_TEST_HPKE_SIGNING_CERTIFICATE="$$(cat /tmp/certificate)"; \
-	docker compose -f ./crates/daphne-server/docker-compose-e2e.yaml up --build --abort-on-container-exit --exit-code-from test
+	docker compose -f ./crates/daphne-server/docker-compose-e2e.yaml up \
+		--no-attach leader_storage \
+		--no-attach helper_storage \
+		--build \
+		--abort-on-container-exit \
+		--exit-code-from test
 
 build_interop:
 	docker build . -f ./interop/Dockerfile.interop_helper --tag daphne-interop
diff --git a/crates/dapf/src/acceptance/mod.rs b/crates/dapf/src/acceptance/mod.rs
index d2852b1..a2ecd0a 100644
--- a/crates/dapf/src/acceptance/mod.rs
+++ b/crates/dapf/src/acceptance/mod.rs
@@ -24,7 +24,6 @@ use crate::{
 use anyhow::{anyhow, bail, Context, Result};
 use async_trait::async_trait;
 use daphne::{
-    auth::BearerToken,
     constants::DapMediaType,
     error::aborts::ProblemDetails,
     hpke::{HpkeConfig, HpkeKemId, HpkeReceiverConfig},
@@ -40,7 +39,7 @@ use daphne::{
     DapQueryConfig, DapTaskConfig, DapTaskParameters, DapVersion, EarlyReportStateConsumed,
     EarlyReportStateInitialized, ReplayProtection,
 };
-use daphne_service_utils::http_headers;
+use daphne_service_utils::{bearer_token::BearerToken, http_headers};
 use futures::{future::OptionFuture, StreamExt, TryStreamExt};
 use prio::codec::{Decode, ParameterizedEncode};
 use prometheus::{Encoder, HistogramVec, IntCounterVec, IntGaugeVec, TextEncoder};
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/Cargo.toml b/crates/daphne-server/Cargo.toml
index efec364..7a4d216 100644
--- a/crates/daphne-server/Cargo.toml
+++ b/crates/daphne-server/Cargo.toml
@@ -16,6 +16,7 @@ description = "Workers backend for Daphne"
 axum = "0.6.0" # held back to use http 0.2
 daphne = { path = "../daphne" }
 daphne-service-utils = { path = "../daphne-service-utils", features = ["durable_requests"] }
+either.workspace = true
 futures.workspace = true
 hex.workspace = true
 http = "0.2" # held back to use http 0.2
@@ -43,13 +44,14 @@ assert_matches.workspace = true
 clap.workspace = true
 config.workspace = true
 daphne = { path = "../daphne", features = ["test-utils"] }
-daphne-service-utils = { path = "../daphne-service-utils", features = ["prometheus"] }
+daphne-service-utils = { path = "../daphne-service-utils", features = ["prometheus", "test-utils"] }
 dhat.workspace = true
 hpke-rs.workspace = true
 paste.workspace = true
 prometheus.workspace = true
 rand.workspace = true
 rcgen.workspace = true
+serde_yaml.workspace = true
 tokio = { workspace = true, features = ["signal"] }
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
 webpki.workspace = true
diff --git a/crates/daphne-server/examples/configuration-helper.toml b/crates/daphne-server/examples/configuration-helper.toml
index 56673c4..7385737 100644
--- a/crates/daphne-server/examples/configuration-helper.toml
+++ b/crates/daphne-server/examples/configuration-helper.toml
@@ -22,13 +22,8 @@ allow_taskprov = true
 default_num_agg_span_shards = 4
 
 [service.taskprov]
+peer_auth.leader.expected_token = "I-am-the-leader" # SECRET
 vdaf_verify_key_init = "b029a72fa327931a5cb643dcadcaafa098fcbfac07d990cb9e7c9a8675fafb18" # SECRET
-leader_auth = """{
-  "bearer_token": "I-am-the-leader"
-}""" # SECRET
-collector_auth = """{
-  "bearer_token": "I-am-the-collector"
-}""" # SECRET
 hpke_collector_config = """{
     "id": 23,
     "kem_id": "p256_hkdf_sha256",
diff --git a/crates/daphne-server/examples/configuration-leader.toml b/crates/daphne-server/examples/configuration-leader.toml
index b9b5257..469bd35 100644
--- a/crates/daphne-server/examples/configuration-leader.toml
+++ b/crates/daphne-server/examples/configuration-leader.toml
@@ -22,13 +22,9 @@ allow_taskprov = true
 default_num_agg_span_shards = 4
 
 [service.taskprov]
+peer_auth.collector.expected_token = "I-am-the-collector" # SECRET
+self_bearer_token = "I-am-the-leader" # SECRET
 vdaf_verify_key_init = "b029a72fa327931a5cb643dcadcaafa098fcbfac07d990cb9e7c9a8675fafb18" # SECRET
-leader_auth = """{
-  "bearer_token": "I-am-the-leader"
-}""" # SECRET
-collector_auth = """{
-  "bearer_token": "I-am-the-collector"
-}""" # SECRET
 hpke_collector_config = """{
     "id": 23,
     "kem_id": "p256_hkdf_sha256",
diff --git a/crates/daphne-server/examples/service.rs b/crates/daphne-server/examples/service.rs
index 0c3c355..94ea9ce 100644
--- a/crates/daphne-server/examples/service.rs
+++ b/crates/daphne-server/examples/service.rs
@@ -8,11 +8,11 @@ use daphne_server::{router, App, StorageProxyConfig};
 use daphne_service_utils::{
     config::DaphneServiceConfig, metrics::DaphnePromServiceMetrics, DapRole,
 };
-use serde::Deserialize;
+use serde::{Deserialize, Serialize};
 use tracing_subscriber::EnvFilter;
 use url::Url;
 
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Serialize, Deserialize)]
 struct Config {
     service: DaphneServiceConfig,
     port: u16,
@@ -98,7 +98,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Sync + Send>> {
 
     // Parse the configuration from the command line arguments.
     let config = Config::try_from(Args::parse())?;
-    println!("starting service with config:\n{config:#?}");
+    println!(
+        "starting service with config:\n{}",
+        serde_yaml::to_string(&config).unwrap()
+    );
 
     // Create a new prometheus registry where metrics will be registered and measured
     let registry = prometheus::Registry::new();
diff --git a/crates/daphne-server/src/lib.rs b/crates/daphne-server/src/lib.rs
index d992c96..75ae51a 100644
--- a/crates/daphne-server/src/lib.rs
+++ b/crates/daphne-server/src/lib.rs
@@ -5,12 +5,19 @@ use std::sync::Arc;
 
 use daphne::{
     audit_log::{AuditLog, NoopAuditLog},
-    auth::BearerToken,
-    roles::leader::in_memory_leader::InMemoryLeaderState,
-    DapError,
+    fatal_error,
+    messages::{Base64Encode, TaskId},
+    roles::{leader::in_memory_leader::InMemoryLeaderState, DapAggregator},
+    DapError, DapSender,
 };
-use daphne_service_utils::{config::DaphneServiceConfig, metrics::DaphneServiceMetrics};
+use daphne_service_utils::{
+    bearer_token::BearerToken,
+    config::{DaphneServiceConfig, PeerBearerToken},
+    metrics::DaphneServiceMetrics,
+};
+use either::Either::{self, Left, Right};
 use futures::lock::Mutex;
+use roles::BearerTokens;
 use serde::{Deserialize, Serialize};
 use storage_proxy_connection::{kv, Do, Kv};
 use tokio::sync::RwLock;
@@ -38,7 +45,11 @@ mod storage_proxy_connection;
 /// use url::Url;
 /// use daphne::{DapGlobalConfig, hpke::HpkeKemId, DapVersion};
 /// use daphne_server::{App, router, StorageProxyConfig};
-/// use daphne_service_utils::{config::DaphneServiceConfig, DapRole, metrics::DaphnePromServiceMetrics};
+/// use daphne_service_utils::{
+///     config::DaphneServiceConfig,
+///     DapRole,
+///     metrics::DaphnePromServiceMetrics
+/// };
 ///
 /// let storage_proxy_settings = StorageProxyConfig {
 ///     url: Url::parse("http://example.com").unwrap(),
@@ -51,7 +62,7 @@ mod storage_proxy_connection;
 ///     min_batch_interval_start: 259_200,
 ///     max_batch_interval_end: 259_200,
 ///     supported_hpke_kems: vec![HpkeKemId::X25519HkdfSha256],
-///     allow_taskprov: true,
+///     allow_taskprov: false,
 ///     default_num_agg_span_shards: NonZeroUsize::new(2).unwrap(),
 /// };
 /// let service_config = DaphneServiceConfig {
@@ -92,6 +103,7 @@ pub struct StorageProxyConfig {
     pub auth_token: BearerToken,
 }
 
+#[axum::async_trait]
 impl router::DaphneService for App {
     fn server_metrics(&self) -> &dyn DaphneServiceMetrics {
         &*self.metrics
@@ -100,6 +112,72 @@ impl router::DaphneService for App {
     fn signing_key(&self) -> Option<&p256::ecdsa::SigningKey> {
         self.service_config.signing_key.as_ref()
     }
+
+    async fn check_bearer_token(
+        &self,
+        presented_token: &BearerToken,
+        sender: DapSender,
+        task_id: TaskId,
+        is_taskprov: bool,
+    ) -> Result<(), Either<String, DapError>> {
+        let reject = |extra_args| {
+            Err(Left(format!(
+                "the indicated bearer token is incorrect for the {sender:?} {extra_args}",
+            )))
+        };
+        if let Some(taskprov) = self
+            .service_config
+            .taskprov
+            .as_ref()
+            // we only use taskprov auth if it's allowed by config and if the request is using taskprov
+            .filter(|_| self.service_config.global.allow_taskprov && is_taskprov)
+        {
+            match (&taskprov.peer_auth, sender) {
+                (PeerBearerToken::Leader { expected_token }, DapSender::Leader)
+                | (PeerBearerToken::Collector { expected_token }, DapSender::Collector)
+                    if expected_token == presented_token =>
+                {
+                    Ok(())
+                }
+                (PeerBearerToken::Leader { .. }, DapSender::Collector) => Err(Right(fatal_error!(
+                    err = "expected a leader sender but got a collector sender"
+                ))),
+                (PeerBearerToken::Collector { .. }, DapSender::Leader) => Err(Right(fatal_error!(
+                    err = "expected a collector sender but got a leader sender"
+                ))),
+                _ => reject(format_args!("using taskprov")),
+            }
+        } else if self
+            .bearer_tokens()
+            .matches(sender, task_id, presented_token)
+            .await
+            .map_err(|e| {
+                Right(fatal_error!(
+                    err = ?e,
+                    "internal error occurred while running authentication"
+                ))
+            })?
+        {
+            Ok(())
+        } else {
+            reject(format_args!("with task_id {}", task_id.to_base64url()))
+        }
+    }
+
+    async fn is_using_taskprov(&self, req: &daphne::DapRequestMeta) -> Result<bool, DapError> {
+        if req.taskprov.is_some() {
+            Ok(true)
+        } else if self
+            .get_task_config_for(&req.task_id)
+            .await?
+            .is_some_and(|task_config| task_config.method_is_taskprov())
+        {
+            tracing::warn!("Client referencing a taskprov task id without taskprov advertisement");
+            Ok(true)
+        } else {
+            Ok(false)
+        }
+    }
 }
 
 impl App {
@@ -137,4 +215,8 @@ impl App {
     pub(crate) fn kv(&self) -> Kv<'_> {
         Kv::new(&self.storage_proxy_config, &self.http, &self.cache)
     }
+
+    pub(crate) fn bearer_tokens(&self) -> BearerTokens<'_> {
+        BearerTokens::from(Kv::new(&self.storage_proxy_config, &self.http, &self.cache))
+    }
 }
diff --git a/crates/daphne-server/src/roles/aggregator.rs b/crates/daphne-server/src/roles/aggregator.rs
index f748dee..78a53e0 100644
--- a/crates/daphne-server/src/roles/aggregator.rs
+++ b/crates/daphne-server/src/roles/aggregator.rs
@@ -1,12 +1,11 @@
 // Copyright (c) 2024 Cloudflare, Inc. All rights reserved.
 // SPDX-License-Identifier: BSD-3-Clause
 
-use std::{borrow::Cow, future::ready, num::NonZeroUsize, ops::Range, time::SystemTime};
+use std::{future::ready, num::NonZeroUsize, ops::Range, time::SystemTime};
 
 use axum::async_trait;
 use daphne::{
     audit_log::AuditLog,
-    auth::{BearerToken, BearerTokenProvider},
     error::DapAbort,
     fatal_error,
     hpke::{HpkeConfig, HpkeDecrypter, HpkeProvider},
@@ -14,13 +13,11 @@ 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::{
-    auth::DaphneAuth,
-    durable_requests::bindings::{
-        self, AggregateStoreMergeOptions, AggregateStoreMergeReq, AggregateStoreMergeResp,
-    },
+use daphne_service_utils::durable_requests::bindings::{
+    self, AggregateStoreMergeOptions, AggregateStoreMergeReq, AggregateStoreMergeResp,
 };
 use futures::{future::try_join_all, StreamExt, TryStreamExt};
 use mappable_rc::Marc;
@@ -32,7 +29,7 @@ use crate::{
 };
 
 #[async_trait]
-impl DapAggregator<DaphneAuth> for crate::App {
+impl DapAggregator for crate::App {
     #[tracing::instrument(skip(self, task_config, agg_share_span))]
     async fn try_put_agg_share_span(
         &self,
@@ -151,54 +148,6 @@ impl DapAggregator<DaphneAuth> for crate::App {
     where
         Self: 'a;
 
-    async fn unauthorized_reason(
-        &self,
-        task_config: &DapTaskConfig,
-        req: &DapRequest<DaphneAuth>,
-    ) -> Result<Option<String>, DapError> {
-        let mut authorized = false;
-
-        let Some(ref sender_auth) = req.sender_auth else {
-            return Ok(Some("Missing authorization.".into()));
-        };
-
-        // If a bearer token is present, verify that it can be used to authorize the request.
-        if sender_auth.bearer_token.is_some() {
-            if let Some(unauthorized_reason) =
-                self.bearer_token_authorized(task_config, req).await?
-            {
-                return Ok(Some(unauthorized_reason));
-            }
-            authorized = true;
-        }
-
-        // If a TLS client certificate is present verify that it is valid.
-        if let Some(ref cf_tls_client_auth) = sender_auth.cf_tls_client_auth {
-            // TODO(cjpatton) Add support for TLS client authentication for non-Taskprov tasks.
-            let Some(ref _taskprov_config) = self.service_config.taskprov else {
-                return Ok(Some(
-                    "TLS client authentication is currently only supported with Taskprov.".into(),
-                ));
-            };
-
-            // Check that that the certificate is valid. This is indicated by literal "SUCCESS".
-            if cf_tls_client_auth.verified != "SUCCESS" {
-                return Ok(Some(format!(
-                    "Invalid TLS certificate ({}).",
-                    cf_tls_client_auth.verified
-                )));
-            }
-
-            authorized = true;
-        }
-
-        if authorized {
-            Ok(None)
-        } else {
-            Ok(Some("No suitable authorization method was found.".into()))
-        }
-    }
-
     async fn get_global_config(&self) -> Result<DapGlobalConfig, DapError> {
         let mut global_config = self.service_config.global.clone();
 
@@ -278,10 +227,10 @@ impl DapAggregator<DaphneAuth> for crate::App {
 
     async fn taskprov_put(
         &self,
-        req: &DapRequest<DaphneAuth>,
+        req: &DapRequestMeta,
         task_config: DapTaskConfig,
     ) -> Result<(), DapError> {
-        let task_id = req.task_id().map_err(DapError::Abort)?;
+        let task_id = &req.task_id;
         let expiration_time = task_config.not_after;
 
         if self.service_config.role.is_leader() || req.taskprov.is_none() {
@@ -524,56 +473,3 @@ impl HpkeDecrypter for crate::App {
             .ok_or(DapError::Transition(TransitionFailure::HpkeUnknownConfigId))?
     }
 }
-
-#[async_trait]
-impl BearerTokenProvider for crate::App {
-    type WrappedBearerToken<'a> = Cow<'a,  BearerToken>
-        where Self: 'a;
-
-    async fn get_leader_bearer_token_for<'s>(
-        &'s self,
-        task_id: &'s TaskId,
-        task_config: &DapTaskConfig,
-    ) -> std::result::Result<Option<Self::WrappedBearerToken<'s>>, DapError> {
-        if self.service_config.global.allow_taskprov && task_config.method_is_taskprov() {
-            if let Some(bearer_token) = self
-                .service_config
-                .taskprov
-                .as_ref()
-                .and_then(|c| c.leader_auth.bearer_token.as_ref())
-            {
-                return Ok(Some(Cow::Borrowed(bearer_token)));
-            }
-        }
-
-        self.kv()
-            .get_cloned::<kv::prefix::LeaderBearerToken>(task_id, &KvGetOptions::default())
-            .await
-            .map_err(|e| fatal_error!(err = ?e, "failed to get the leader bearer token"))
-            .map(|r| r.map(Cow::Owned))
-    }
-
-    async fn get_collector_bearer_token_for<'s>(
-        &'s self,
-        task_id: &'s TaskId,
-        task_config: &DapTaskConfig,
-    ) -> std::result::Result<Option<Self::WrappedBearerToken<'s>>, DapError> {
-        if self.service_config.global.allow_taskprov && task_config.method_is_taskprov() {
-            if let Some(bearer_token) = self.service_config.taskprov.as_ref().and_then(|c| {
-                c.collector_auth
-                    .as_ref()
-                    .expect("collector auth method not set")
-                    .bearer_token
-                    .as_ref()
-            }) {
-                return Ok(Some(Cow::Borrowed(bearer_token)));
-            }
-        }
-
-        self.kv()
-            .get_cloned::<kv::prefix::CollectorBearerToken>(task_id, &KvGetOptions::default())
-            .await
-            .map_err(|e| fatal_error!(err = ?e, "failed to get the collector bearer token"))
-            .map(|r| r.map(Cow::Owned))
-    }
-}
diff --git a/crates/daphne-server/src/roles/helper.rs b/crates/daphne-server/src/roles/helper.rs
index 5d07a16..f312c74 100644
--- a/crates/daphne-server/src/roles/helper.rs
+++ b/crates/daphne-server/src/roles/helper.rs
@@ -3,7 +3,6 @@
 
 use axum::async_trait;
 use daphne::roles::DapHelper;
-use daphne_service_utils::auth::DaphneAuth;
 
 #[async_trait]
-impl DapHelper<DaphneAuth> for crate::App {}
+impl DapHelper for crate::App {}
diff --git a/crates/daphne-server/src/roles/leader.rs b/crates/daphne-server/src/roles/leader.rs
index 66b58b9..eff3d49 100644
--- a/crates/daphne-server/src/roles/leader.rs
+++ b/crates/daphne-server/src/roles/leader.rs
@@ -3,46 +3,25 @@
 
 #![allow(unused_variables)]
 
-use std::time::Instant;
+use std::{borrow::Cow, time::Instant};
 
 use axum::{async_trait, http::Method};
 use daphne::{
-    auth::BearerTokenProvider,
     constants::DapMediaType,
     error::DapAbort,
     fatal_error,
     messages::{BatchId, BatchSelector, Collection, CollectionJobId, Report, TaskId},
-    roles::{leader::WorkItem, DapAggregator, DapAuthorizedSender, DapLeader},
-    DapAggregationParam, DapCollectionJob, DapError, DapRequest, DapResponse, DapTaskConfig,
+    roles::{leader::WorkItem, DapAggregator, DapLeader},
+    DapAggregationParam, DapCollectionJob, DapError, DapRequest, DapResponse, DapVersion,
 };
-use daphne_service_utils::{auth::DaphneAuth, http_headers};
+use daphne_service_utils::http_headers;
+use http::StatusCode;
+use prio::codec::ParameterizedEncode;
 use tracing::{error, info};
 use url::Url;
 
 #[async_trait]
-impl DapAuthorizedSender<DaphneAuth> for crate::App {
-    async fn authorize(
-        &self,
-        task_id: &TaskId,
-        task_config: &DapTaskConfig,
-        media_type: &DapMediaType,
-        _payload: &[u8],
-    ) -> Result<DaphneAuth, DapError> {
-        Ok(DaphneAuth {
-            bearer_token: Some(
-                self.authorize_with_bearer_token(task_id, task_config, media_type)
-                    .await?
-                    .into_owned(),
-            ),
-            // TODO Consider adding support for authorizing the request with TLS client
-            // certificates: https://developers.cloudflare.com/workers/runtime-apis/mtls/
-            cf_tls_client_auth: None,
-        })
-    }
-}
-
-#[async_trait]
-impl DapLeader<DaphneAuth> for crate::App {
+impl DapLeader for crate::App {
     async fn put_report(&self, report: &Report, task_id: &TaskId) -> Result<(), DapError> {
         let task_config = self
             .get_task_config_for(task_id)
@@ -121,32 +100,32 @@ impl DapLeader<DaphneAuth> for crate::App {
         self.test_leader_state.lock().await.enqueue_work(items)
     }
 
-    async fn send_http_post(
-        &self,
-        req: DapRequest<DaphneAuth>,
-        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<DaphneAuth>,
-        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<DaphneAuth>,
+        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
             .and_then(|mt| mt.as_str_for_version(req.version))
@@ -165,18 +144,43 @@ impl crate::App {
                 .map_err(|e| fatal_error!(err = ?e, "failed to construct content-type header"))?,
         );
 
-        if let Some(bearer_token) = req.sender_auth.and_then(|auth| auth.bearer_token) {
-            headers.insert(
-                HeaderName::from_static(http_headers::DAP_AUTH_TOKEN),
-                HeaderValue::from_str(bearer_token.as_ref()).map_err(|e| {
-                    fatal_error!(
-                        err = ?e,
-                        "failed to construct {} header",
-                        http_headers::DAP_AUTH_TOKEN
-                    )
-                })?,
-            );
-        }
+        let bearer_token = if req.taskprov.is_some() {
+            if let Some(bearer_token) = self
+                .service_config
+                .taskprov
+                .as_ref()
+                .and_then(|t| t.self_bearer_token.as_ref())
+            {
+                Cow::Borrowed(bearer_token)
+            } else {
+                return Err(DapError::Abort(DapAbort::UnauthorizedRequest {
+                    detail: format!(
+                        "taskprov authentication not setup for authentication with peer at {url}",
+                    ),
+                    task_id: req.task_id,
+                }));
+            }
+        } else if let Some(bearer_token) = self
+            .bearer_tokens()
+            .get(daphne::DapSender::Leader, req.task_id)
+            .await
+            .map_err(|e| fatal_error!(err = ?e, "failed to get leader bearer token"))?
+        {
+            Cow::Owned(bearer_token)
+        } else {
+            return Err(DapError::Abort(DapAbort::UnauthorizedRequest {
+                detail: format!(
+                    "no suitable authentication method found for authenticating with peer at {url}",
+                ),
+                task_id: req.task_id,
+            }));
+        };
+
+        headers.insert(
+            HeaderName::from_static(http_headers::DAP_AUTH_TOKEN),
+            HeaderValue::from_str(bearer_token.as_str())
+                .map_err(|e| fatal_error!(err = ?e, "failed to construct authentication header"))?,
+        );
 
         if let Some(taskprov_advertisement) = req.taskprov.as_deref() {
             headers.insert(
@@ -187,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();
@@ -208,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
@@ -218,25 +227,35 @@ impl crate::App {
                 .to_vec();
 
             Ok(DapResponse {
-                version: req.version,
+                version: meta.version,
                 payload,
                 media_type,
             })
         } else {
             error!("{}: request failed: {:?}", url, reqwest_resp);
-            if status == 400 {
-                if let Some(content_type) =
-                    reqwest_resp.headers().get(reqwest::header::CONTENT_TYPE)
-                {
-                    if content_type == "application/problem+json" {
-                        error!(
-                            "Problem details: {}",
-                            reqwest_resp.text().await.map_err(
-                                |e| fatal_error!(err = ?e, "failed to read body of helper error response")
-                            )?
-                        );
+            match status {
+                StatusCode::BAD_REQUEST => {
+                    if let Some(content_type) =
+                        reqwest_resp.headers().get(reqwest::header::CONTENT_TYPE)
+                    {
+                        if content_type == "application/problem+json" {
+                            error!(
+                                "Problem details: {}",
+                                reqwest_resp.text().await.map_err(
+                                    |e| fatal_error!(err = ?e, "failed to read body of helper error response")
+                                )?
+                            );
+                        }
+                    }
+                }
+                StatusCode::UNAUTHORIZED => {
+                    return Err(DapAbort::UnauthorizedRequest {
+                        detail: format!("helper at {url} didn't authorize our request"),
+                        task_id: meta.task_id,
                     }
+                    .into())
                 }
+                _ => {}
             }
             Err(fatal_error!(err = "request aborted by peer"))
         }
diff --git a/crates/daphne-server/src/roles/mod.rs b/crates/daphne-server/src/roles/mod.rs
index 0d460be..b0566d0 100644
--- a/crates/daphne-server/src/roles/mod.rs
+++ b/crates/daphne-server/src/roles/mod.rs
@@ -1,9 +1,13 @@
 // Copyright (c) 2024 Cloudflare, Inc. All rights reserved.
 // SPDX-License-Identifier: BSD-3-Clause
 
-use daphne::ReplayProtection;
+use daphne::{messages::TaskId, DapSender, ReplayProtection};
+use daphne_service_utils::bearer_token::BearerToken;
 
-use crate::storage_proxy_connection::kv::{self, Kv, KvGetOptions};
+use crate::storage_proxy_connection::{
+    self,
+    kv::{self, Kv, KvGetOptions},
+};
 
 mod aggregator;
 mod helper;
@@ -32,18 +36,85 @@ pub async fn fetch_replay_protection_override(kv: Kv<'_>) -> ReplayProtection {
     }
 }
 
+/// Bearer token for for tasks configured manually or via the [ppm-dap-interop-test][interop] draft.
+///
+/// [interop]: https://divergentdave.github.io/draft-dcook-ppm-dap-interop-test-design/draft-dcook-ppm-dap-interop-test-design.html
+pub(crate) struct BearerTokens<'s> {
+    kv: kv::Kv<'s>,
+}
+
+impl<'s> From<kv::Kv<'s>> for BearerTokens<'s> {
+    fn from(kv: kv::Kv<'s>) -> Self {
+        Self { kv }
+    }
+}
+
+impl BearerTokens<'_> {
+    #[cfg(feature = "test-utils")]
+    pub async fn put_if_not_exists(
+        &self,
+        role: DapSender,
+        task_id: TaskId,
+        token: BearerToken,
+    ) -> Result<Option<BearerToken>, storage_proxy_connection::Error> {
+        self.kv
+            .put_if_not_exists::<kv::prefix::KvBearerToken>(&(role, task_id).into(), token)
+            .await
+    }
+
+    /// Checks if a presented token matches the expected token of a task.
+    ///
+    /// # Returns
+    ///
+    /// - `Ok(true)` if the task exists and the token matches
+    /// - `Ok(false)` if the task doesn't exist or the token doesn't match
+    /// - `Err(error)` if any io error occurs while fetching
+    pub async fn matches(
+        &self,
+        role: DapSender,
+        task_id: TaskId,
+        token: &BearerToken,
+    ) -> Result<bool, storage_proxy_connection::Error> {
+        self.kv
+            .peek::<kv::prefix::KvBearerToken, _, _>(
+                &(role, task_id).into(),
+                &kv::KvGetOptions {
+                    cache_not_found: false,
+                },
+                |stored_token| stored_token == token,
+            )
+            .await
+            .map(|s| s.is_some_and(|matches| matches))
+    }
+
+    pub async fn get(
+        &self,
+        role: DapSender,
+        task_id: TaskId,
+    ) -> Result<Option<BearerToken>, storage_proxy_connection::Error> {
+        self.kv
+            .get_cloned::<kv::prefix::KvBearerToken>(
+                &(role, task_id).into(),
+                &kv::KvGetOptions {
+                    cache_not_found: false,
+                },
+            )
+            .await
+    }
+}
+
 #[cfg(feature = "test-utils")]
 mod test_utils {
     use daphne::{
-        auth::BearerToken,
         fatal_error,
         hpke::{HpkeConfig, HpkeReceiverConfig},
         messages::decode_base64url_vec,
         roles::DapAggregator,
         vdaf::{Prio3Config, VdafConfig},
-        DapError, DapQueryConfig, DapTaskConfig, DapVersion,
+        DapError, DapQueryConfig, DapSender, DapTaskConfig, DapVersion,
     };
     use daphne_service_utils::{
+        bearer_token::BearerToken,
         test_route_types::{InternalTestAddTask, InternalTestEndpointForTask},
         DapRole,
     };
@@ -61,7 +132,7 @@ mod test_utils {
 
             self.http
                 .delete(self.storage_proxy_config.url.join(PURGE_STORAGE).unwrap())
-                .bearer_auth(&self.storage_proxy_config.auth_token)
+                .bearer_auth(self.storage_proxy_config.auth_token.as_str())
                 .send()
                 .await
                 .map_err(
@@ -77,7 +148,7 @@ mod test_utils {
             use daphne_service_utils::durable_requests::STORAGE_READY;
             self.http
                 .get(self.storage_proxy_config.url.join(STORAGE_READY).unwrap())
-                .bearer_auth(&self.storage_proxy_config.auth_token)
+                .bearer_auth(self.storage_proxy_config.auth_token.as_str())
                 .send()
                 .await
                 .map_err(|e| fatal_error!(err = ?e, "failed to send ready check request to storage proxy"))?
@@ -155,30 +226,26 @@ mod test_utils {
             // Leader authentication token.
             let token = BearerToken::from(cmd.leader_authentication_token);
             if self
-                .kv()
-                .put_if_not_exists::<kv::prefix::LeaderBearerToken>(&cmd.task_id, token)
+                .bearer_tokens()
+                .put_if_not_exists(DapSender::Leader, cmd.task_id, token)
                 .await
-                .map_err(|e| fatal_error!(err = ?e, "failed to fetch leader bearer token"))?
-                .is_some()
+                .is_err()
             {
                 return Err(fatal_error!(
                     err = "command failed: token already exists for the given task and bearer role (leader)",
                     task_id = %cmd.task_id,
                 ));
-            }
+            };
 
             // Collector authentication token.
             match (cmd.role, cmd.collector_authentication_token) {
                 (DapRole::Leader, Some(token_string)) => {
                     let token = BearerToken::from(token_string);
                     if self
-                        .kv()
-                        .put_if_not_exists::<kv::prefix::CollectorBearerToken>(&cmd.task_id, token)
+                        .bearer_tokens()
+                        .put_if_not_exists(DapSender::Collector, cmd.task_id, token)
                         .await
-                        .map_err(
-                            |e| fatal_error!(err = ?e, "failed to put collector bearer token"),
-                        )?
-                        .is_some()
+                        .is_err()
                     {
                         return Err(fatal_error!(err = format!(
                             "command failed: token already exists for the given task ({}) and bearer role (collector)",
diff --git a/crates/daphne-server/src/router/aggregator.rs b/crates/daphne-server/src/router/aggregator.rs
index 9b66d15..4cd33bf 100644
--- a/crates/daphne-server/src/router/aggregator.rs
+++ b/crates/daphne-server/src/router/aggregator.rs
@@ -5,7 +5,7 @@ use std::sync::Arc;
 
 use axum::{
     body::HttpBody,
-    extract::{Query, State},
+    extract::{Path, Query, State},
     response::{AppendHeaders, IntoResponse},
     routing::get,
 };
@@ -14,17 +14,17 @@ use daphne::{
     fatal_error,
     messages::{encode_base64url, TaskId},
     roles::{aggregator, DapAggregator},
-    DapError, DapResponse,
+    DapError, DapResponse, DapVersion,
 };
-use daphne_service_utils::{auth::DaphneAuth, http_headers};
+use daphne_service_utils::http_headers;
 use p256::ecdsa::{signature::Signer, Signature, SigningKey};
 use serde::Deserialize;
 
-use super::{AxumDapResponse, DapRequestExtractor, DaphneService};
+use super::{AxumDapResponse, DaphneService};
 
 pub fn add_aggregator_routes<A, B>(router: super::Router<A, B>) -> super::Router<A, B>
 where
-    A: DapAggregator<DaphneAuth> + DaphneService + Send + Sync + 'static,
+    A: DapAggregator + DaphneService + Send + Sync + 'static,
     B: Send + HttpBody + 'static,
     B::Data: Send,
     B::Error: Send + Sync,
@@ -38,16 +38,16 @@ struct QueryTaskId {
     task_id: Option<TaskId>,
 }
 
-#[tracing::instrument(skip(app, req), fields(version = ?req.version))]
+#[tracing::instrument(skip(app), fields(version, task_id))]
 async fn hpke_config<A>(
     State(app): State<Arc<A>>,
     Query(QueryTaskId { task_id }): Query<QueryTaskId>,
-    DapRequestExtractor(req): DapRequestExtractor,
+    Path(version): Path<DapVersion>,
 ) -> impl IntoResponse
 where
-    A: DapAggregator<DaphneAuth> + DaphneService,
+    A: DapAggregator + DaphneService,
 {
-    match aggregator::handle_hpke_config_req(&*app, &req, task_id).await {
+    match aggregator::handle_hpke_config_req(&*app, version, task_id).await {
         Ok(resp) => match app.signing_key().map(|k| sign_dap_response(k, &resp)) {
             None => AxumDapResponse::new_success(resp, app.server_metrics()).into_response(),
             Some(Ok(signed)) => (
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 087c537..8dc0937 100644
--- a/crates/daphne-server/src/router/helper.rs
+++ b/crates/daphne-server/src/router/helper.rs
@@ -9,11 +9,9 @@ use axum::{
     routing::{post, put},
 };
 use daphne::{
-    constants::DapMediaType,
-    error::DapAbort,
+    messages::{AggregateShareReq, AggregationJobInitReq},
     roles::{helper, DapHelper},
 };
-use daphne_service_utils::auth::DaphneAuth;
 use http::StatusCode;
 
 use crate::{roles::fetch_replay_protection_override, App};
@@ -38,52 +36,40 @@ where
     skip_all,
     fields(
         media_type = ?req.media_type,
-        task_id = ?req.task_id().ok(),
-        version = ?req.version
+        task_id = ?req.task_id,
+        version = ?req.version,
     )
 )]
 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(
     skip_all,
     fields(
         media_type = ?req.media_type,
-        task_id = ?req.task_id().ok(),
-        version = ?req.version
+        task_id = ?req.task_id,
+        version = ?req.version,
     )
 )]
 async fn agg_share<A>(
     State(app): State<Arc<A>>,
-    DapRequestExtractor(req): DapRequestExtractor,
+    DapRequestExtractor(req): DapRequestExtractor<AggregateShareReq>,
 ) -> AxumDapResponse
 where
-    A: DapHelper<DaphneAuth> + DaphneService + Send + Sync,
+    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 95208ef..2fa2e45 100644
--- a/crates/daphne-server/src/router/leader.rs
+++ b/crates/daphne-server/src/router/leader.rs
@@ -13,17 +13,19 @@ use axum::{
 use daphne::{
     constants::DapMediaType,
     error::DapAbort,
+    messages,
     roles::leader::{self, DapLeader},
     DapError, DapVersion,
 };
-use daphne_service_utils::auth::DaphneAuth;
 use prio::codec::ParameterizedEncode;
 
-use super::{AxumDapResponse, DapRequestExtractor, DaphneService};
+use super::{
+    AxumDapResponse, DapRequestExtractor, DaphneService, UnauthenticatedDapRequestExtractor,
+};
 
 pub(super) fn add_leader_routes<A, B>(router: super::Router<A, B>) -> super::Router<A, B>
 where
-    A: DapLeader<DaphneAuth> + DaphneService + Send + Sync + 'static,
+    A: DapLeader + DaphneService + Send + Sync + 'static,
     B: Send + HttpBody + 'static,
     B::Data: Send,
     B::Error: Send + Sync,
@@ -43,18 +45,18 @@ where
 #[tracing::instrument(
     skip_all,
     fields(
-        task_id = ?req.task_id().ok(),
-        version = ?req.version
+        task_id = ?req.task_id,
+        version = ?req.version,
     )
 )]
 async fn upload<A>(
     State(app): State<Arc<A>>,
-    DapRequestExtractor(req): DapRequestExtractor,
+    UnauthenticatedDapRequestExtractor(req): UnauthenticatedDapRequestExtractor<messages::Report>,
 ) -> Response
 where
-    A: DapLeader<DaphneAuth> + DaphneService + Send + Sync,
+    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(),
     }
@@ -63,16 +65,16 @@ where
 #[tracing::instrument(
     skip_all,
     fields(
-        task_id = ?req.task_id().ok(),
-        version = ?req.version
+        task_id = ?req.task_id,
+        version = ?req.version,
     )
 )]
 async fn get_collect_uri<A>(
     State(app): State<Arc<A>>,
-    DapRequestExtractor(req): DapRequestExtractor,
+    DapRequestExtractor(req): DapRequestExtractor<messages::CollectionReq>,
 ) -> Response
 where
-    A: DapLeader<DaphneAuth> + DaphneService + Send + Sync,
+    A: DapLeader + DaphneService + Send + Sync,
 {
     match (leader::handle_coll_job_req(&*app, &req).await, req.version) {
         (Ok(collect_uri), DapVersion::Draft09 | DapVersion::Latest) => {
@@ -85,26 +87,22 @@ where
 #[tracing::instrument(
     skip_all,
     fields(
-        task_id = ?req.task_id().ok(),
-        version = ?req.version
+        task_id = ?req.task_id,
+        version = ?req.version,
     )
 )]
 async fn collect<A>(
     State(app): State<Arc<A>>,
-    DapRequestExtractor(req): DapRequestExtractor,
+    DapRequestExtractor(req): DapRequestExtractor<()>,
 ) -> Response
 where
-    A: DapLeader<DaphneAuth> + DaphneService + Send + Sync,
+    A: DapLeader + DaphneService + Send + Sync,
 {
-    let task_id = match req.task_id() {
-        Ok(id) => id,
-        Err(e) => return AxumDapResponse::new_error(e, app.server_metrics()).into_response(),
-    };
     let collect_id = match req.collection_job_id() {
         Ok(id) => id,
         Err(e) => return AxumDapResponse::new_error(e, app.server_metrics()).into_response(),
     };
-    match app.poll_collect_job(task_id, collect_id).await {
+    match app.poll_collect_job(&req.task_id, collect_id).await {
         Ok(daphne::DapCollectionJob::Done(collect_resp)) => AxumDapResponse::new_success(
             daphne::DapResponse {
                 version: req.version,
diff --git a/crates/daphne-server/src/router/mod.rs b/crates/daphne-server/src/router/mod.rs
index bcf616c..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,36 +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::{
-    auth::BearerToken,
-    constants::DapMediaType,
-    error::DapAbort,
-    fatal_error,
-    messages::{AggregationJobId, CollectionJobId, TaskId},
-    DapError, DapRequest, DapResource, DapResponse, DapVersion,
-};
-use daphne_service_utils::{
-    auth::{DaphneAuth, TlsClientAuth},
-    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::Request;
-use serde::Deserialize;
 
 use crate::App;
+use extractor::{DapRequestExtractor, UnauthenticatedDapRequestExtractor};
 
 type Router<A, B> = axum::Router<Arc<A>, B>;
 
 /// Capabilities necessary when running a native daphne service.
+#[axum::async_trait]
 pub trait DaphneService {
     /// The service metrics
     fn server_metrics(&self) -> &dyn DaphneServiceMetrics;
@@ -47,11 +40,30 @@ pub trait DaphneService {
     fn signing_key(&self) -> Option<&p256::ecdsa::SigningKey> {
         None
     }
+
+    /// Checks if a bearer token is accepted.
+    ///
+    /// # Errors
+    ///
+    /// Returns an either:
+    /// - left: error message with the reason why the token wasn't accepted.
+    /// - right: an internal error that made checking the token impossible.
+    async fn check_bearer_token(
+        &self,
+        presented_token: &BearerToken,
+        sender: DapSender,
+        task_id: TaskId,
+        is_taskprov: bool,
+    ) -> Result<(), Either<String, DapError>>;
+
+    /// Checks if this request intends to use taskprov.
+    async fn is_using_taskprov(&self, req: &DapRequestMeta) -> Result<bool, DapError>;
 }
 
+#[axum::async_trait]
 impl<S> DaphneService for Arc<S>
 where
-    S: DaphneService,
+    S: DaphneService + Send + Sync,
 {
     fn server_metrics(&self) -> &dyn DaphneServiceMetrics {
         S::server_metrics(&**self)
@@ -60,6 +72,20 @@ where
     fn signing_key(&self) -> Option<&p256::ecdsa::SigningKey> {
         S::signing_key(&**self)
     }
+
+    async fn check_bearer_token(
+        &self,
+        presented_token: &BearerToken,
+        sender: DapSender,
+        task_id: TaskId,
+        is_taskprov: bool,
+    ) -> Result<(), Either<String, DapError>> {
+        S::check_bearer_token(&**self, presented_token, sender, task_id, is_taskprov).await
+    }
+
+    async fn is_using_taskprov(&self, req: &DapRequestMeta) -> Result<bool, DapError> {
+        S::is_using_taskprov(&**self, req).await
+    }
 }
 
 pub fn new<B>(role: DapRole, aggregator: App) -> axum::Router<(), B>
@@ -153,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
@@ -208,291 +235,3 @@ impl IntoResponse for AxumDapResponse {
         self.0
     }
 }
-
-/// An axum extractor capable of parsing a [`DapRequest`].
-#[derive(Debug)]
-struct DapRequestExtractor(pub DapRequest<DaphneAuth>);
-
-#[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> {
-        #[derive(Debug, Deserialize)]
-        #[serde(deny_unknown_fields)]
-        struct PathParams {
-            version: DapVersion,
-            #[serde(default, with = "daphne::messages::base64url_option")]
-            task_id: Option<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 extract_header_as_string = |header: &'static str| -> Option<String> {
-            parts
-                .headers
-                .get(header)?
-                .to_str()
-                .ok()
-                .map(ToString::to_string)
-        };
-
-        let sender_auth = DaphneAuth {
-            bearer_token: extract_header_as_string(http_headers::DAP_AUTH_TOKEN)
-                .map(BearerToken::from),
-
-            cf_tls_client_auth: extract_header_as_string("X-Client-Cert-Verified")
-                .map(|verified| TlsClientAuth { verified }),
-        };
-
-        if sender_auth.bearer_token.is_some() {
-            state
-                .server_metrics()
-                .auth_method_inc(metrics::AuthMethod::BearerToken);
-        }
-        if sender_auth.cf_tls_client_auth.is_some() {
-            state
-                .server_metrics()
-                .auth_method_inc(metrics::AuthMethod::TlsClientAuth);
-        }
-
-        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(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)
-        };
-
-        Ok(DapRequestExtractor(DapRequest {
-            version,
-            task_id,
-            resource,
-            payload: payload.to_vec(),
-            media_type,
-            sender_auth: Some(sender_auth),
-            taskprov,
-        }))
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use std::sync::{Arc, OnceLock};
-
-    use axum::{
-        body::{Body, HttpBody},
-        extract::State,
-        http::{header::CONTENT_TYPE, Request, StatusCode},
-        response::IntoResponse,
-        routing::get,
-        Router,
-    };
-    use daphne::{
-        async_test_version, async_test_versions,
-        messages::{AggregationJobId, Base64Encode, TaskId},
-        DapRequest, DapResource, DapVersion,
-    };
-    use daphne_service_utils::{auth::DaphneAuth, metrics::DaphnePromServiceMetrics};
-    use futures::future::BoxFuture;
-    use rand::{thread_rng, Rng};
-    use tokio::sync::mpsc::{self, Sender};
-    use tower::ServiceExt;
-
-    use super::DapRequestExtractor;
-
-    /// Return a function that will parse a request using the [`DapRequestExtractor`] and return
-    /// the parsed request.
-    ///
-    /// The possible request URIs that are supported by this parser are:
-    ///  - `/:version/parse-version`
-    ///  - `/:version/:task_id/parse-task-id`
-    ///  - `/:version/:agg_job_id/parse-agg-job-id`
-    ///  - `/:version/:collect_job_id/parse-collect-job-id`
-    fn test_router<B>() -> impl FnOnce(Request<B>) -> BoxFuture<'static, DapRequest<DaphneAuth>>
-    where
-        B: Send + Sync + 'static + HttpBody,
-        B::Data: Send,
-        B::Error: Send + Sync + std::error::Error,
-    {
-        type Channel = Sender<DapRequest<DaphneAuth>>;
-
-        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 handler(
-            State(ch): State<Arc<Channel>>,
-            DapRequestExtractor(req): DapRequestExtractor,
-        ) -> impl IntoResponse {
-            ch.send(req).await.unwrap();
-        }
-
-        let (tx, mut rx) = mpsc::channel(1);
-
-        let router = Router::new()
-            .route("/:version/parse-version", get(handler))
-            .route("/:version/:task_id/parse-task-id", get(handler))
-            .route("/:version/:agg_job_id/parse-agg-job-id", get(handler))
-            .route(
-                "/:version/:collect_job_id/parse-collect-job-id",
-                get(handler),
-            )
-            .with_state(Arc::new(tx));
-
-        move |req| {
-            Box::pin(async move {
-                let resp = match router.oneshot(req).await {
-                    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()
-                            )
-                        )
-                    }
-                    code => assert_eq!(code, StatusCode::OK),
-                }
-
-                rx.recv().await.unwrap()
-            })
-        }
-    }
-
-    async fn parse_version(version: DapVersion) {
-        let test = test_router();
-
-        let req = test(
-            Request::builder()
-                .uri(format!("/{version}/parse-version"))
-                .body(Body::empty())
-                .unwrap(),
-        )
-        .await;
-
-        assert_eq!(req.version, version);
-    }
-
-    async_test_versions! { parse_version }
-
-    async fn parse_task_id(version: DapVersion) {
-        let test = test_router();
-
-        let task_id = TaskId(thread_rng().gen());
-
-        let req = test(
-            Request::builder()
-                .uri(format!(
-                    "/{version}/{}/parse-task-id",
-                    task_id.to_base64url()
-                ))
-                .body(Body::empty())
-                .unwrap(),
-        )
-        .await;
-
-        assert_eq!(req.task_id, Some(task_id));
-    }
-
-    async_test_versions! { parse_task_id }
-
-    async fn parse_agg_job_id(version: DapVersion) {
-        let test = test_router();
-
-        let agg_job_id = AggregationJobId(thread_rng().gen());
-
-        let req = test(
-            Request::builder()
-                .uri(format!(
-                    "/{version}/{}/parse-agg-job-id",
-                    agg_job_id.to_base64url()
-                ))
-                .header(CONTENT_TYPE, "application/dap-aggregation-job-init-req")
-                .body(Body::empty())
-                .unwrap(),
-        )
-        .await;
-
-        assert_eq!(req.resource, DapResource::AggregationJob(agg_job_id));
-    }
-
-    async_test_version! { parse_agg_job_id, Draft09 }
-    async_test_version! { parse_agg_job_id, Latest }
-}
diff --git a/crates/daphne-server/src/storage_proxy_connection/kv/mod.rs b/crates/daphne-server/src/storage_proxy_connection/kv/mod.rs
index 387d8e6..1cbd349 100644
--- a/crates/daphne-server/src/storage_proxy_connection/kv/mod.rs
+++ b/crates/daphne-server/src/storage_proxy_connection/kv/mod.rs
@@ -33,10 +33,16 @@ pub trait KvPrefix {
 }
 
 pub mod prefix {
-    use std::{fmt::Display, marker::PhantomData};
-
-    use daphne::{auth::BearerToken, messages::TaskId, taskprov, DapTaskConfig, DapVersion};
-    use daphne_service_utils::config::HpkeRecieverConfigList;
+    use std::{
+        fmt::{self, Display},
+        marker::PhantomData,
+    };
+
+    use daphne::{
+        messages::{Base64Encode, TaskId},
+        taskprov, DapSender, DapTaskConfig, DapVersion,
+    };
+    use daphne_service_utils::{bearer_token::BearerToken, config::HpkeRecieverConfigList};
     use serde::{de::DeserializeOwned, Serialize};
 
     use super::KvPrefix;
@@ -98,20 +104,32 @@ pub mod prefix {
         type Value = HpkeRecieverConfigList;
     }
 
-    pub struct LeaderBearerToken();
-    impl KvPrefix for LeaderBearerToken {
-        const PREFIX: &'static str = "bearer_token/leader/task";
+    pub struct KvBearerToken();
+    impl KvPrefix for KvBearerToken {
+        const PREFIX: &'static str = "bearer_token";
 
-        type Key = TaskId;
+        type Key = KvBearerTokenKey;
         type Value = BearerToken;
     }
 
-    pub struct CollectorBearerToken();
-    impl KvPrefix for CollectorBearerToken {
-        const PREFIX: &'static str = "bearer_token/collector/task";
-
-        type Key = TaskId;
-        type Value = BearerToken;
+    #[derive(Debug)]
+    pub struct KvBearerTokenKey(DapSender, TaskId);
+    impl From<(DapSender, TaskId)> for KvBearerTokenKey {
+        fn from((s, t): (DapSender, TaskId)) -> Self {
+            Self(s, t)
+        }
+    }
+    impl fmt::Display for KvBearerTokenKey {
+        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+            let Self(sender, task_id) = self;
+            let task_id = task_id.to_base64url();
+            match sender {
+                DapSender::Client => write!(f, "client/task/{task_id}"),
+                DapSender::Collector => write!(f, "collector/task/{task_id}"),
+                DapSender::Helper => write!(f, "helper/task/{task_id}"),
+                DapSender::Leader => write!(f, "leader/task/{task_id}"),
+            }
+        }
     }
 }
 
@@ -233,7 +251,7 @@ impl<'h> Kv<'h> {
             let resp = self
                 .http
                 .get(self.config.url.join(&key).unwrap())
-                .bearer_auth(&self.config.auth_token)
+                .bearer_auth(self.config.auth_token.as_str())
                 .send()
                 .await?;
             if resp.status() == StatusCode::NOT_FOUND {
@@ -275,7 +293,7 @@ impl<'h> Kv<'h> {
         let mut request = self
             .http
             .post(self.config.url.join(&key).unwrap())
-            .bearer_auth(&self.config.auth_token)
+            .bearer_auth(self.config.auth_token.as_str())
             .body(serde_json::to_vec(&value).unwrap());
 
         if let Some(expiration) = expiration {
@@ -336,7 +354,7 @@ impl<'h> Kv<'h> {
         let mut request = self
             .http
             .put(self.config.url.join(&key).unwrap())
-            .bearer_auth(&self.config.auth_token)
+            .bearer_auth(self.config.auth_token.as_str())
             .body(serde_json::to_vec(&value).unwrap());
 
         if let Some(expiration) = expiration {
diff --git a/crates/daphne-server/src/storage_proxy_connection/mod.rs b/crates/daphne-server/src/storage_proxy_connection/mod.rs
index 8798345..19ed049 100644
--- a/crates/daphne-server/src/storage_proxy_connection/mod.rs
+++ b/crates/daphne-server/src/storage_proxy_connection/mod.rs
@@ -79,7 +79,7 @@ impl<'d, B: DurableMethod + Debug, P: AsRef<[u8]>> RequestBuilder<'d, B, P> {
             .http
             .post(url)
             .body(self.request.into_bytes())
-            .bearer_auth(&self.durable.config.auth_token)
+            .bearer_auth(self.durable.config.auth_token.as_str())
             .send()
             .await?;
 
diff --git a/crates/daphne-service-utils/src/auth.rs b/crates/daphne-service-utils/src/auth.rs
deleted file mode 100644
index 4a7898a..0000000
--- a/crates/daphne-service-utils/src/auth.rs
+++ /dev/null
@@ -1,124 +0,0 @@
-// Copyright (c) 2023 Cloudflare, Inc. All rights reserved.
-// SPDX-License-Identifier: BSD-3-Clause
-
-//! Authorization methods for Daphne-Worker.
-
-use std::fmt::Debug;
-
-use daphne::auth::BearerToken;
-use serde::{Deserialize, Serialize};
-
-#[derive(PartialEq, Eq)]
-pub struct TlsClientAuth {
-    pub verified: String,
-}
-
-/// HTTP client authorization for Daphne-Worker.
-///
-/// Multiple authorization methods can be configured. The sender may present multiple authorization
-/// methods; the request is authorized if validation of all presented methods succeed. If an
-/// authorization method is presented, but the server is not configured to validate it, then
-/// validation of that method will fail.
-//
-// TODO(cjpatton) Add an authorization method for Cloudflare Access
-// (https://www.cloudflare.com/products/zero-trust/access/). This allows us to delegate access
-// control to that service; Daphne-Worker would just need to verify that Access granted access.
-#[derive(PartialEq)]
-pub struct DaphneAuth {
-    /// Bearer token, expected to appear in the
-    /// [`DAP_AUTH_TOKEN`](crate::http_headers::DAP_AUTH_TOKEN) header.
-    pub bearer_token: Option<BearerToken>,
-
-    /// TLS client authentication. The client uses a certificate when establishing the TLS
-    /// connection. This authorization method is Cloudflare-specific: Verifying the certificate
-    /// itself is handled by the process that invoked this Worker. The customer zone is also
-    /// expected to be configured to require mutual TLS for the route on which this Worker is
-    /// listening.
-    ///
-    /// When this authorization method is used, we verify that a certificate was presented and was
-    /// successfully verified by the TLS server.
-    ///
-    /// # Caveats
-    ///
-    /// * For now, only the Helper supports TLS client auth; the Leader still expects a bearer
-    ///   token to be configured for the task.
-    ///
-    /// * For now, TLS client auth is only enabled if the taskprov extension is configured.
-    ///   Enabling this feature for other tasks will require a bit plumbing.
-    pub cf_tls_client_auth: Option<TlsClientAuth>,
-}
-
-// Custom debug implementation to avoid exposing sensitive information.
-impl Debug for DaphneAuth {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        // pattern match on self to get a compiler error if the struct changes
-        let Self {
-            bearer_token,
-            cf_tls_client_auth,
-        } = self;
-
-        fn opt_to_str<T>(o: &Option<T>) -> &dyn Debug {
-            if o.is_some() {
-                &"is-present"
-            } else {
-                &"is-missing"
-            }
-        }
-
-        f.debug_struct("DaphneAuth")
-            .field("bearer_token", opt_to_str(bearer_token))
-            .field("cf_tls_client_auth", opt_to_str(cf_tls_client_auth))
-            .finish()
-    }
-}
-
-// TODO(mendess): remove this implementation. Implementations of AsRef should never panic
-impl AsRef<BearerToken> for DaphneAuth {
-    fn as_ref(&self) -> &BearerToken {
-        if let Some(ref bearer_token) = self.bearer_token {
-            bearer_token
-        } else {
-            // We would only try this method if we previously resolved to use a bearer token for
-            // authorization.
-            unreachable!("no bearer token provided by sender")
-        }
-    }
-}
-
-#[derive(Debug, Deserialize, Serialize, Clone)]
-pub struct DaphneWorkerAuthMethod {
-    /// Expected bearer token.
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub bearer_token: Option<BearerToken>,
-}
-
-// TODO(mendess): remove this implementation. Implementations of AsRef should never panic
-impl AsRef<BearerToken> for DaphneWorkerAuthMethod {
-    fn as_ref(&self) -> &BearerToken {
-        if let Some(ref bearer_token) = self.bearer_token {
-            bearer_token
-        } else {
-            // We would only try this method if we previously resolved to use a bearer token for
-            // authorization.
-            unreachable!("no bearer token provided by sender")
-        }
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use super::{BearerToken, DaphneWorkerAuthMethod};
-
-    #[test]
-    fn daphne_worker_auth_method_json_serialization() {
-        let daphne_worker_auth_method: DaphneWorkerAuthMethod =
-            serde_json::from_str(r#"{"bearer_token":"the bearer token"}"#).unwrap();
-        assert_eq!(
-            daphne_worker_auth_method.bearer_token,
-            Some(BearerToken::from("the bearer token".to_string()))
-        );
-
-        let daphne_worker_auth_method: DaphneWorkerAuthMethod = serde_json::from_str("{}").unwrap();
-        assert!(daphne_worker_auth_method.bearer_token.is_none());
-    }
-}
diff --git a/crates/daphne-service-utils/src/bearer_token.rs b/crates/daphne-service-utils/src/bearer_token.rs
new file mode 100644
index 0000000..d0c8d33
--- /dev/null
+++ b/crates/daphne-service-utils/src/bearer_token.rs
@@ -0,0 +1,57 @@
+// Copyright (c) 2022 Cloudflare, Inc. All rights reserved.
+// SPDX-License-Identifier: BSD-3-Clause
+
+//! DAP request authorization.
+
+use core::fmt;
+
+use daphne::messages::constant_time_eq;
+use serde::{Deserialize, Serialize};
+
+/// A bearer token used for authorizing DAP requests.
+#[derive(Clone, Deserialize, Serialize, Eq)]
+#[serde(transparent)]
+pub struct BearerToken {
+    raw: String,
+}
+
+impl BearerToken {
+    pub fn as_str(&self) -> &str {
+        self.raw.as_str()
+    }
+}
+
+impl fmt::Debug for BearerToken {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        #[cfg(feature = "test-utils")]
+        {
+            write!(f, "BearerToken({})", self.raw)
+        }
+        #[cfg(not(feature = "test-utils"))]
+        write!(f, "BearerToken(REDACTED)")
+    }
+}
+
+impl AsRef<str> for BearerToken {
+    fn as_ref(&self) -> &str {
+        self.as_str()
+    }
+}
+
+impl PartialEq for BearerToken {
+    fn eq(&self, other: &Self) -> bool {
+        constant_time_eq(self.raw.as_bytes(), other.raw.as_bytes())
+    }
+}
+
+impl From<String> for BearerToken {
+    fn from(raw: String) -> Self {
+        Self { raw }
+    }
+}
+
+impl From<&str> for BearerToken {
+    fn from(raw: &str) -> Self {
+        Self::from(raw.to_string())
+    }
+}
diff --git a/crates/daphne-service-utils/src/config.rs b/crates/daphne-service-utils/src/config.rs
index 4b729c3..17e69ee 100644
--- a/crates/daphne-service-utils/src/config.rs
+++ b/crates/daphne-service-utils/src/config.rs
@@ -9,7 +9,7 @@ use p256::ecdsa::SigningKey;
 use serde::{Deserialize, Serialize};
 use url::Url;
 
-use crate::{auth::DaphneWorkerAuthMethod, DapRole};
+use crate::{bearer_token::BearerToken, DapRole};
 
 /// draft-wang-ppm-dap-taskprov: Long-lived parameters for the taskprov extension.
 #[derive(Serialize, Deserialize, Debug, Clone)]
@@ -22,13 +22,22 @@ pub struct TaskprovConfig {
     #[serde(with = "hex")]
     pub vdaf_verify_key_init: [u8; 32],
 
-    /// Leader, Helper: Method for authorizing Leader requests.
-    #[serde(with = "from_raw_string")]
-    pub leader_auth: DaphneWorkerAuthMethod,
+    /// Peer's bearer token.
+    pub peer_auth: PeerBearerToken,
 
-    /// Leader: Method for authorizing Collector requests.
-    #[serde(default, with = "from_raw_string")]
-    pub collector_auth: Option<DaphneWorkerAuthMethod>,
+    /// Bearer token used when trying to communicate with an aggregator using taskprov.
+    #[serde(default)]
+    pub self_bearer_token: Option<BearerToken>,
+}
+
+/// Peer authentication tokens for incomming requests. Different roles have different peers.
+/// - Helpers have a Leader peer.
+/// - Leaders have a Collector peer.
+#[derive(Serialize, Deserialize, Debug, Clone)]
+#[serde(rename_all = "lowercase")]
+pub enum PeerBearerToken {
+    Leader { expected_token: BearerToken },
+    Collector { expected_token: BearerToken },
 }
 
 pub type HpkeRecieverConfigList = Vec<HpkeReceiverConfig>;
diff --git a/crates/daphne-service-utils/src/lib.rs b/crates/daphne-service-utils/src/lib.rs
index e311bd8..240df21 100644
--- a/crates/daphne-service-utils/src/lib.rs
+++ b/crates/daphne-service-utils/src/lib.rs
@@ -5,12 +5,13 @@ use std::str::FromStr;
 
 use serde::{Deserialize, Serialize};
 
-pub mod auth;
+pub mod bearer_token;
 pub mod config;
 #[cfg(feature = "durable_requests")]
 pub mod durable_requests;
 pub mod http_headers;
 pub mod metrics;
+#[cfg(feature = "test-utils")]
 pub mod test_route_types;
 
 // the generated code expects this module to be defined at the root of the library.
diff --git a/crates/daphne/src/auth.rs b/crates/daphne/src/auth.rs
deleted file mode 100644
index 46caca4..0000000
--- a/crates/daphne/src/auth.rs
+++ /dev/null
@@ -1,170 +0,0 @@
-// Copyright (c) 2022 Cloudflare, Inc. All rights reserved.
-// SPDX-License-Identifier: BSD-3-Clause
-
-//! DAP request authorization.
-
-use std::fmt::Display;
-
-use crate::{
-    constants::DapMediaType,
-    fatal_error,
-    messages::{constant_time_eq, TaskId},
-    DapError, DapRequest, DapSender, DapTaskConfig,
-};
-use async_trait::async_trait;
-use serde::{Deserialize, Serialize};
-
-/// A bearer token used for authorizing DAP requests.
-#[derive(Clone, Debug, Deserialize, Serialize)]
-#[cfg_attr(any(test, feature = "test-utils"), derive(deepsize::DeepSizeOf))]
-#[serde(transparent)]
-pub struct BearerToken {
-    raw: String,
-}
-
-impl BearerToken {
-    pub fn as_str(&self) -> &str {
-        self.raw.as_str()
-    }
-}
-
-impl AsRef<str> for BearerToken {
-    fn as_ref(&self) -> &str {
-        self.as_str()
-    }
-}
-
-impl PartialEq for BearerToken {
-    fn eq(&self, other: &Self) -> bool {
-        constant_time_eq(self.raw.as_bytes(), other.raw.as_bytes())
-    }
-}
-
-impl From<String> for BearerToken {
-    fn from(raw: String) -> Self {
-        Self { raw }
-    }
-}
-
-impl From<&str> for BearerToken {
-    fn from(raw: &str) -> Self {
-        Self::from(raw.to_string())
-    }
-}
-
-impl AsRef<BearerToken> for BearerToken {
-    fn as_ref(&self) -> &Self {
-        self
-    }
-}
-
-impl Display for BearerToken {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.write_str(self.as_str())
-    }
-}
-
-/// A source of bearer tokens used for authorizing DAP requests.
-#[async_trait]
-pub trait BearerTokenProvider {
-    /// A reference to a bearer token owned by the provider.
-    type WrappedBearerToken<'a>: AsRef<BearerToken> + Send
-    where
-        Self: 'a;
-
-    /// Fetch the Leader's bearer token for the given task, if the task is recognized.
-    async fn get_leader_bearer_token_for<'s>(
-        &'s self,
-        task_id: &'s TaskId,
-        task_config: &DapTaskConfig,
-    ) -> Result<Option<Self::WrappedBearerToken<'s>>, DapError>;
-
-    /// Fetch the Collector's bearer token for the given task, if the task is recognized.
-    async fn get_collector_bearer_token_for<'s>(
-        &'s self,
-        task_id: &'s TaskId,
-        task_config: &DapTaskConfig,
-    ) -> Result<Option<Self::WrappedBearerToken<'s>>, DapError>;
-
-    /// Return a bearer token that can be used to authorize a request with the given task ID and
-    /// media type.
-    async fn authorize_with_bearer_token<'s>(
-        &'s self,
-        task_id: &'s TaskId,
-        task_config: &DapTaskConfig,
-        media_type: &DapMediaType,
-    ) -> Result<Self::WrappedBearerToken<'s>, DapError> {
-        if matches!(media_type.sender(), DapSender::Leader) {
-            let token = self
-                .get_leader_bearer_token_for(task_id, task_config)
-                .await?
-                .ok_or_else(|| {
-                    fatal_error!(err = "attempted to authorize request with unknown task ID",)
-                })?;
-            return Ok(token);
-        }
-
-        Err(fatal_error!(
-            err = "attempted to authorize request of type",
-            ?media_type,
-        ))
-    }
-
-    /// Check that the bearer token carried by a request can be used to authorize that request.
-    ///
-    /// Return `None` if the request is authorized. Otherwise return `Some(reason)`, where `reason`
-    /// is the reason for the failure.
-    async fn bearer_token_authorized<T: AsRef<BearerToken> + Send + Sync>(
-        &self,
-        task_config: &DapTaskConfig,
-        req: &DapRequest<T>,
-    ) -> Result<Option<String>, DapError> {
-        if req.task_id.is_none() {
-            // Can't authorize request with missing task ID.
-            return Ok(Some(
-                "Cannot authorize request with missing task ID.".into(),
-            ));
-        }
-        let task_id = req.task_id.as_ref().unwrap();
-
-        // TODO spec: Decide whether to check that the bearer token has the right format, say,
-        // following RFC 6750, Section 2.1. Note that we would also need to replace `From<String>
-        // for BearerToken` with `TryFrom<String>` so that a `DapError` can be returned if the
-        // token is not formatted properly.
-        if matches!(req.sender(), Some(DapSender::Leader)) {
-            if let Some(ref got) = req.sender_auth {
-                if let Some(expected) = self
-                    .get_leader_bearer_token_for(task_id, task_config)
-                    .await?
-                {
-                    return Ok(if got.as_ref() == expected.as_ref() {
-                        None
-                    } else {
-                        Some("The indicated bearer token is incorrect for the Leader.".into())
-                    });
-                }
-            }
-        }
-
-        if matches!(req.sender(), Some(DapSender::Collector)) {
-            if let Some(ref got) = req.sender_auth {
-                if let Some(expected) = self
-                    .get_collector_bearer_token_for(task_id, task_config)
-                    .await?
-                {
-                    return Ok(if got.as_ref() == expected.as_ref() {
-                        None
-                    } else {
-                        Some("The indicated bearer token is incorrect for the Collector.".into())
-                    });
-                }
-            }
-        }
-
-        // Deny request with unhandled or unknown media type.
-        Ok(Some(format!(
-            "Cannot resolve sender due to unexpected media type ({:?}).",
-            req.media_type
-        )))
-    }
-}
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 94391a2..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<S>(req: &DapRequest<S>, 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 06e20b5..33b38e9 100644
--- a/crates/daphne/src/lib.rs
+++ b/crates/daphne/src/lib.rs
@@ -41,7 +41,6 @@
 //!     > requests to a collect job URI whose results have been removed.
 
 pub mod audit_log;
-pub mod auth;
 pub mod constants;
 pub mod error;
 pub mod hpke;
@@ -59,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};
@@ -92,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,
 };
@@ -1049,107 +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<S> {
-    /// 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. This field is optional, since some
-    /// requests may apply to all tasks, e.g., the request for the HPKE configuration.
-    pub task_id: Option<TaskId>,
-
-    /// The resource with which this request is associated.
-    pub resource: DapResource,
-
-    /// Request payload.
-    pub payload: Vec<u8>,
-
-    /// Sender authorization, e.g., a bearer token.
-    pub sender_auth: Option<S>,
-
-    /// taskprov: The task advertisement, sent in the `dap-taskprov` header.
-    pub taskprov: Option<String>,
-}
-
-#[cfg(test)]
-impl<S> Default for DapRequest<S> {
-    fn default() -> Self {
-        Self {
-            version: DapVersion::Draft09,
-            media_type: None,
-            task_id: Default::default(),
-            resource: Default::default(),
-            payload: Default::default(),
-            sender_auth: Default::default(),
-            taskprov: Default::default(),
-        }
-    }
-}
-
-impl<S> DapRequest<S> {
-    /// Return the task ID, handling a missing ID as a user error.
-    pub fn task_id(&self) -> Result<&TaskId, DapAbort> {
-        if let Some(ref id) = self.task_id {
-            Ok(id)
-        } else {
-            // Handle missing task ID as a bad request. The task ID is normally conveyed by the
-            // request path; if missing at this point, it is because it was missing or couldn't be
-            // parsed from the request path.
-            Err(DapAbort::BadRequest("missing or malformed task ID".into()))
-        }
-    }
-
-    /// 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 edc4d39..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,
+    DapRequestMeta, DapResponse, DapTaskConfig, DapVersion,
 };
 
 /// Report initializer. Used by a DAP Aggregator [`DapAggregator`] when initializing an aggregation
@@ -45,23 +45,12 @@ pub enum MergeAggShareError {
 
 /// DAP Aggregator functionality.
 #[async_trait]
-pub trait DapAggregator<S: Sync>: HpkeProvider + DapReportInitializer + Sized {
+pub trait DapAggregator: HpkeProvider + DapReportInitializer + Sized {
     /// A refernce to a task configuration stored by the Aggregator.
     type WrappedDapTaskConfig<'a>: AsRef<DapTaskConfig> + Send
     where
         Self: 'a;
 
-    /// Decide whether the given DAP request is authorized.
-    ///
-    /// If the return value is `None`, then the request is authorized. If the return value is
-    /// `Some(reason)`, then the request is denied and `reason` conveys details about how the
-    /// decision was reached.
-    async fn unauthorized_reason(
-        &self,
-        task_config: &DapTaskConfig,
-        req: &DapRequest<S>,
-    ) -> Result<Option<String>, DapError>;
-
     /// Look up the DAP global configuration.
     async fn get_global_config(&self) -> Result<DapGlobalConfig, DapError>;
 
@@ -95,7 +84,7 @@ pub trait DapAggregator<S: Sync>: HpkeProvider + DapReportInitializer + Sized {
     /// nothing.
     async fn taskprov_put(
         &self,
-        req: &DapRequest<S>,
+        req: &DapRequestMeta,
         task_config: DapTaskConfig,
     ) -> Result<(), DapError>;
 
@@ -164,19 +153,18 @@ pub trait DapAggregator<S: Sync>: HpkeProvider + DapReportInitializer + Sized {
 }
 
 /// Handle request for the Aggregator's HPKE configuration.
-pub async fn handle_hpke_config_req<S, A>(
+pub async fn handle_hpke_config_req<A>(
     aggregator: &A,
-    req: &DapRequest<S>,
+    version: DapVersion,
     task_id: Option<TaskId>,
 ) -> Result<DapResponse, DapError>
 where
-    S: Sync,
-    A: DapAggregator<S>,
+    A: DapAggregator,
 {
     let metrics = aggregator.metrics();
 
     let hpke_config = aggregator
-        .get_hpke_config_for(req.version, task_id.as_ref())
+        .get_hpke_config_for(version, task_id.as_ref())
         .await?;
 
     if let Some(task_id) = task_id {
@@ -186,10 +174,8 @@ where
             .ok_or(DapAbort::UnrecognizedTask { task_id })?;
 
         // Check whether the DAP version in the request matches the task config.
-        if task_config.as_ref().version != req.version {
-            return Err(
-                DapAbort::version_mismatch(req.version, task_config.as_ref().version).into(),
-            );
+        if task_config.as_ref().version != version {
+            return Err(DapAbort::version_mismatch(version, task_config.as_ref().version).into());
         }
     }
 
@@ -202,7 +188,7 @@ where
 
     metrics.inbound_req_inc(DaphneRequestType::HpkeConfig);
     Ok(DapResponse {
-        version: req.version,
+        version,
         media_type: DapMediaType::HpkeConfigList,
         payload,
     })
diff --git a/crates/daphne/src/roles/helper.rs b/crates/daphne/src/roles/helper.rs
index 74835bd..3d839e4 100644
--- a/crates/daphne/src/roles/helper.rs
+++ b/crates/daphne/src/roles/helper.rs
@@ -5,9 +5,8 @@ use std::{collections::HashMap, sync::Once};
 
 use async_trait::async_trait;
 use prio::codec::{Encode, ParameterizedDecode};
-use tracing::error;
 
-use super::{check_batch, check_request_content_type, resolve_taskprov, DapAggregator};
+use super::{check_batch, resolve_taskprov, DapAggregator};
 use crate::{
     constants::DapMediaType,
     error::DapAbort,
@@ -24,42 +23,30 @@ use crate::{
 
 /// DAP Helper functionality.
 #[async_trait]
-pub trait DapHelper<S: Sync>: DapAggregator<S> {}
+pub trait DapHelper: DapAggregator {}
 
-pub async fn handle_agg_job_init_req<'req, S: Sync, A: DapHelper<S>>(
+pub async fn handle_agg_job_init_req<A: DapHelper>(
     aggregator: &A,
-    req: &'req DapRequest<S>,
+    req: DapRequest<AggregationJobInitReq>,
     replay_protection: ReplayProtection,
 ) -> Result<DapResponse, DapError> {
     let global_config = aggregator.get_global_config().await?;
-    let task_id = req.task_id()?;
+    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
-        .get_task_config_for(task_id)
+        .get_task_config_for(&task_id)
         .await?
-        .ok_or(DapAbort::UnrecognizedTask { task_id: *task_id })?;
+        .ok_or(DapAbort::UnrecognizedTask { task_id })?;
     let task_config = wrapped_task_config.as_ref();
 
-    if let Some(reason) = aggregator.unauthorized_reason(task_config, req).await? {
-        error!("aborted unauthorized collect request: {reason}");
-        return Err(DapAbort::UnauthorizedRequest {
-            detail: reason,
-            task_id: *task_id,
-        }
-        .into());
-    }
-
     let DapResource::AggregationJob(_agg_job_id) = req.resource else {
         return Err(DapAbort::BadRequest("missing aggregation job ID".to_string()).into());
     };
@@ -71,19 +58,20 @@ pub async fn handle_agg_job_init_req<'req, S: Sync, A: DapHelper<S>>(
 
     // Ensure we know which batch the request pertains to.
     check_part_batch(
-        task_id,
+        &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,
+            &task_id,
+            req.payload,
             replay_protection,
         )
         .await?;
@@ -91,7 +79,7 @@ pub async fn handle_agg_job_init_req<'req, S: Sync, A: DapHelper<S>>(
     let agg_job_resp = {
         let agg_job_resp = finish_agg_job_and_aggregate(
             aggregator,
-            task_id,
+            &task_id,
             task_config,
             &part_batch_sel,
             &initialized_reports,
@@ -105,7 +93,7 @@ pub async fn handle_agg_job_init_req<'req, S: Sync, A: DapHelper<S>>(
     };
 
     aggregator.audit_log().on_aggregation_job(
-        task_id,
+        &task_id,
         task_config,
         agg_job_resp.transitions.len() as u64,
         0, /* vdaf step */
@@ -113,104 +101,76 @@ pub async fn handle_agg_job_init_req<'req, S: Sync, A: DapHelper<S>>(
 
     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, S: Sync, A: DapHelper<S>>(
-    aggregator: &A,
-    req: &DapRequest<S>,
-    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, S: Sync, A: DapHelper<S>>(
+pub async fn handle_agg_share_req<'req, A: DapHelper>(
     aggregator: &A,
-    req: &DapRequest<S>,
+    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)?;
+    let task_id = req.task_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 wrapped_task_config = aggregator
-        .get_task_config_for(req.task_id()?)
+        .get_task_config_for(&task_id)
         .await?
-        .ok_or(DapAbort::UnrecognizedTask { task_id: *task_id })?;
+        .ok_or(DapAbort::UnrecognizedTask { task_id })?;
     let task_config = wrapped_task_config.as_ref();
 
-    if let Some(reason) = aggregator.unauthorized_reason(task_config, req).await? {
-        error!("aborted unauthorized collect request: {reason}");
-        return Err(DapAbort::UnauthorizedRequest {
-            detail: reason,
-            task_id: *task_id,
-        }
-        .into());
-    }
-
     // Check whether the DAP version in the request matches the task config.
     if task_config.version != req.version {
         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)
-            .map_err(|e| DapAbort::from_codec_error(e, *task_id))?;
+        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
     // collected batches.
     check_batch(
         aggregator,
         task_config,
-        task_id,
-        &agg_share_req.batch_sel.clone().into(),
-        &agg_share_req.agg_param,
+        &task_id,
+        &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: *task_id,
+                task_id,
             }.into());
     }
 
     // Check the batch size.
     if !task_config
-        .is_report_count_compatible(task_id, agg_share.report_count)
+        .is_report_count_compatible(&task_id, agg_share.report_count)
         .unwrap_or(false)
     {
         return Err(DapAbort::InvalidBatchSize {
@@ -218,20 +178,20 @@ pub async fn handle_agg_share_req<'req, S: Sync, A: DapHelper<S>>(
                 "Report count ({}) is less than minimum ({})",
                 agg_share.report_count, task_config.min_batch_size
             ),
-            task_id: *task_id,
+            task_id,
         }
         .into());
     }
 
     // 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,
+        &task_id,
+        &req.payload.batch_sel,
         &agg_param,
         &agg_share,
         task_config.version,
@@ -241,7 +201,7 @@ pub async fn handle_agg_share_req<'req, S: Sync, A: DapHelper<S>>(
         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,
@@ -276,8 +236,8 @@ fn check_part_batch(
     Ok(())
 }
 
-async fn finish_agg_job_and_aggregate<S: Sync>(
-    helper: &impl DapHelper<S>,
+async fn finish_agg_job_and_aggregate(
+    helper: &impl DapHelper,
     task_id: &TaskId,
     task_config: &DapTaskConfig,
     part_batch_sel: &PartialBatchSelector,
diff --git a/crates/daphne/src/roles/leader/mod.rs b/crates/daphne/src/roles/leader/mod.rs
index 78e184a..0362cff 100644
--- a/crates/daphne/src/roles/leader/mod.rs
+++ b/crates/daphne/src/roles/leader/mod.rs
@@ -9,13 +9,10 @@ use async_trait::async_trait;
 use futures::future::try_join_all;
 use prio::codec::{Decode, Encode, ParameterizedDecode, ParameterizedEncode};
 use rand::{thread_rng, Rng};
-use tracing::{debug, error};
+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<S: Sync>(
-    role: &impl DapLeader<S>,
+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,16 +67,14 @@ async fn leader_send_http_request<S: Sync>(
         .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: Some(*task_id),
-        resource,
-        sender_auth: Some(
-            role.authorize(task_id, task_config, &req_media_type, &req_data)
-                .await?,
-        ),
+        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 {
@@ -88,19 +86,6 @@ async fn leader_send_http_request<S: Sync>(
     Ok(resp)
 }
 
-/// A party in the DAP protocol who is authorized to send requests to another party.
-#[async_trait]
-pub trait DapAuthorizedSender<S> {
-    /// Add authorization to an outbound DAP request with the given task ID, media type, and payload.
-    async fn authorize(
-        &self,
-        task_id: &TaskId,
-        task_config: &DapTaskConfig,
-        media_type: &DapMediaType,
-        payload: &[u8],
-    ) -> Result<S, DapError>;
-}
-
 /// A work item, either an aggregation job or collection job.
 #[derive(Debug)]
 #[cfg_attr(any(test, feature = "test-utils"), derive(deepsize::DeepSizeOf))]
@@ -130,7 +115,7 @@ impl WorkItem {
 
 /// DAP Leader functionality.
 #[async_trait]
-pub trait DapLeader<S: Sync>: DapAuthorizedSender<S> + DapAggregator<S> {
+pub trait DapLeader: DapAggregator {
     /// Store a report for use later on.
     async fn put_report(&self, report: &Report, task_id: &TaskId) -> Result<(), DapError>;
 
@@ -171,35 +156,40 @@ pub trait DapLeader<S: Sync>: DapAuthorizedSender<S> + DapAggregator<S> {
     ) -> Result<(), DapError>;
 
     /// Send an HTTP POST request.
-    async fn send_http_post(&self, req: DapRequest<S>, 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<S>, 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<S: Sync, A: DapLeader<S>>(
+pub async fn handle_upload_req<A: DapLeader>(
     aggregator: &A,
-    req: &DapRequest<S>,
+    req: DapRequest<Report>,
 ) -> Result<(), DapError> {
     let global_config = aggregator.get_global_config().await?;
     let metrics = aggregator.metrics();
-    let task_id = req.task_id()?;
+    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)
+        .get_task_config_for(&task_id)
         .await?
-        .ok_or(DapAbort::UnrecognizedTask { task_id: *task_id })?;
+        .ok_or(DapAbort::UnrecognizedTask { task_id })?;
 
     // Check whether the DAP version in the request matches the task config.
     if task_config.as_ref().version != req.version {
@@ -212,14 +202,14 @@ pub async fn handle_upload_req<S: Sync, A: DapLeader<S>>(
                 "expected exactly two encrypted input shares; got {}",
                 report.encrypted_input_shares.len()
             ),
-            task_id: *task_id,
+            task_id,
         }
         .into());
     }
 
     // Check that the indicated HpkeConfig is present.
     if !aggregator
-        .can_hpke_decrypt(req.task_id()?, report.encrypted_input_shares[0].config_id)
+        .can_hpke_decrypt(&task_id, report.encrypted_input_shares[0].config_id)
         .await?
     {
         return Err(DapAbort::ReportRejected {
@@ -247,7 +237,7 @@ pub async fn handle_upload_req<S: Sync, A: DapLeader<S>>(
     // 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, req.task_id()?).await?;
+    aggregator.put_report(report, &task_id).await?;
 
     metrics.inbound_req_inc(DaphneRequestType::Upload);
     Ok(())
@@ -255,44 +245,31 @@ pub async fn handle_upload_req<S: Sync, A: DapLeader<S>>(
 
 /// Handle a collect job from the Collector. The response is the URI that the Collector will
 /// poll later on to get the collection.
-pub async fn handle_coll_job_req<S: Sync, A: DapLeader<S>>(
+pub async fn handle_coll_job_req<A: DapLeader>(
     aggregator: &A,
-    req: &DapRequest<S>,
+    req: &DapRequest<CollectionReq>,
 ) -> Result<Url, 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()?;
+    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?;
+        resolve_taskprov(aggregator, &task_id, req, &global_config).await?;
     }
 
-    let task_id = req.task_id()?;
     let wrapped_task_config = aggregator
-        .get_task_config_for(task_id)
+        .get_task_config_for(&task_id)
         .await?
-        .ok_or(DapAbort::UnrecognizedTask { task_id: *task_id })?;
+        .ok_or(DapAbort::UnrecognizedTask { task_id })?;
     let task_config = wrapped_task_config.as_ref();
 
-    if let Some(reason) = aggregator.unauthorized_reason(task_config, req).await? {
-        error!("aborted unauthorized collect request: {reason}");
-        return Err(DapAbort::UnauthorizedRequest {
-            detail: reason,
-            task_id: *task_id,
-        }
-        .into());
-    }
-
-    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)
-            .map_err(|e| DapAbort::from_codec_error(e, *task_id))?;
+            .map_err(|e| DapAbort::from_codec_error(e, task_id))?;
 
     // Check whether the DAP version in the request matches the task config.
     if task_config.version != req.version {
@@ -304,7 +281,7 @@ pub async fn handle_coll_job_req<S: Sync, A: DapLeader<S>>(
     check_batch(
         aggregator,
         task_config,
-        task_id,
+        &task_id,
         &coll_job_req.query,
         &coll_job_req.agg_param,
         now,
@@ -312,20 +289,18 @@ pub async fn handle_coll_job_req<S: Sync, A: DapLeader<S>>(
     )
     .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 },
         Query::FixedSizeByBatchId { batch_id } => BatchSelector::FixedSizeByBatchId { batch_id },
         Query::FixedSizeCurrentBatch => BatchSelector::FixedSizeByBatchId {
-            batch_id: aggregator.current_batch(task_id).await?,
+            batch_id: aggregator.current_batch(&task_id).await?,
         },
     };
 
     let collect_job_uri = aggregator
-        .init_collect_job(task_id, coll_job_id, batch_sel, agg_param)
+        .init_collect_job(&task_id, coll_job_id, batch_sel, agg_param)
         .await?;
 
     metrics.inbound_req_inc(DaphneRequestType::Collect);
@@ -334,7 +309,7 @@ pub async fn handle_coll_job_req<S: Sync, A: DapLeader<S>>(
 
 /// Run an aggregation job for a set of reports. Return the number of reports that were
 /// aggregated successfully.
-async fn run_agg_job<S: Sync, A: DapLeader<S>>(
+async fn run_agg_job<A: DapLeader>(
     aggregator: &A,
     task_id: &TaskId,
     task_config: &DapTaskConfig,
@@ -380,9 +355,7 @@ async fn run_agg_job<S: Sync, A: DapLeader<S>>(
             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(),
         },
@@ -440,7 +413,7 @@ async fn run_agg_job<S: Sync, A: DapLeader<S>>(
 /// Handle a pending collection job. If the results are ready, then compute the aggregate
 /// results and store them to be retrieved by the Collector later. Returns the number of
 /// reports in the batch.
-async fn run_coll_job<S: Sync, A: DapLeader<S>>(
+async fn run_coll_job<A: DapLeader>(
     aggregator: &A,
     task_id: &TaskId,
     task_config: &DapTaskConfig,
@@ -484,6 +457,8 @@ async fn run_coll_job<S: Sync, A: DapLeader<S>>(
 
     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,
@@ -494,9 +469,7 @@ async fn run_coll_job<S: Sync, A: DapLeader<S>>(
             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,
         },
@@ -533,11 +506,11 @@ async fn run_coll_job<S: Sync, A: DapLeader<S>>(
 
     // 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.
@@ -547,7 +520,7 @@ async fn run_coll_job<S: Sync, A: DapLeader<S>>(
 ///
 /// Collection jobs are processed in order. If a collection job is still pending once processed, it
 /// is pushed to the back of the work queue.
-pub async fn process<S: Sync, A: DapLeader<S>>(
+pub async fn process<A: DapLeader>(
     aggregator: &A,
     host: &str,
     num_items: usize,
diff --git a/crates/daphne/src/roles/mod.rs b/crates/daphne/src/roles/mod.rs
index a8db54e..6a6a638 100644
--- a/crates/daphne/src/roles/mod.rs
+++ b/crates/daphne/src/roles/mod.rs
@@ -8,18 +8,17 @@ 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;
 
 pub use aggregator::{DapAggregator, DapReportInitializer};
 pub use helper::DapHelper;
-pub use leader::{DapAuthorizedSender, DapLeader};
+pub use leader::DapLeader;
 
-async fn check_batch<S: Sync>(
-    agg: &impl DapAggregator<S>,
+async fn check_batch(
+    agg: &impl DapAggregator,
     task_config: &DapTaskConfig,
     task_id: &TaskId,
     query: &Query,
@@ -92,21 +91,10 @@ async fn check_batch<S: Sync>(
     Ok(())
 }
 
-fn check_request_content_type<S>(
-    req: &DapRequest<S>,
-    expected: DapMediaType,
-) -> Result<(), DapAbort> {
-    if req.media_type != Some(expected) {
-        Err(DapAbort::content_type(req, expected))
-    } else {
-        Ok(())
-    }
-}
-
-async fn resolve_taskprov<S: Sync>(
-    agg: &impl DapAggregator<S>,
+async fn resolve_taskprov(
+    agg: &impl DapAggregator,
     task_id: &TaskId,
-    req: &DapRequest<S>,
+    req: &DapRequestMeta,
     global_config: &DapGlobalConfig,
 ) -> Result<(), DapError> {
     if agg.get_task_config_for(task_id).await?.is_some() {
@@ -146,12 +134,11 @@ async fn resolve_taskprov<S: Sync>(
 
 #[cfg(test)]
 mod test {
-    use super::{aggregator, helper, leader, DapAuthorizedSender, DapLeader};
+    use super::{aggregator, helper, leader, DapLeader};
     #[cfg(feature = "experimental")]
     use crate::vdaf::{mastic::MasticWeight, MasticWeightConfig};
     use crate::{
         assert_metrics_include, async_test_versions,
-        auth::BearerToken,
         constants::DapMediaType,
         hpke::{HpkeKemId, HpkeProvider, HpkeReceiverConfig},
         messages::{
@@ -164,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};
@@ -179,8 +166,6 @@ mod test {
     pub(super) struct TestData {
         pub now: Time,
         global_config: DapGlobalConfig,
-        collector_token: BearerToken,
-        taskprov_collector_token: BearerToken,
         pub time_interval_task_id: TaskId,
         pub fixed_size_task_id: TaskId,
         pub expired_task_id: TaskId,
@@ -188,10 +173,8 @@ mod test {
         pub mastic_task_id: TaskId,
         helper_registry: prometheus::Registry,
         tasks: HashMap<TaskId, DapTaskConfig>,
-        pub leader_token: BearerToken,
         collector_hpke_receiver_config: HpkeReceiverConfig,
         taskprov_vdaf_verify_key_init: [u8; 32],
-        taskprov_leader_token: BearerToken,
         leader_registry: prometheus::Registry,
     }
 
@@ -313,14 +296,8 @@ mod test {
                 );
             }
 
-            // Authorization tokens. These are normally chosen at random.
-            let leader_token = BearerToken::from("leader_token");
-            let collector_token = BearerToken::from("collector_token");
-
             // taskprov
             let taskprov_vdaf_verify_key_init = rng.gen::<[u8; 32]>();
-            let taskprov_leader_token = BearerToken::from("taskprov_leader_token");
-            let taskprov_collector_token = BearerToken::from("taskprov_collector_token");
 
             let helper_registry = prometheus::Registry::new_custom(
                 Option::None,
@@ -342,8 +319,6 @@ mod test {
             Self {
                 now,
                 global_config,
-                collector_token,
-                taskprov_collector_token,
                 time_interval_task_id,
                 fixed_size_task_id,
                 expired_task_id,
@@ -351,8 +326,6 @@ mod test {
                 mastic_task_id,
                 helper_registry,
                 tasks,
-                leader_token,
-                taskprov_leader_token,
                 collector_hpke_receiver_config,
                 taskprov_vdaf_verify_key_init,
                 leader_registry,
@@ -366,11 +339,9 @@ mod test {
                     .gen_hpke_receiver_config_list(thread_rng().gen())
                     .expect("failed to generate HPKE receiver config"),
                 self.global_config.clone(),
-                self.leader_token.clone(),
                 self.collector_hpke_receiver_config.config.clone(),
                 &self.helper_registry,
                 self.taskprov_vdaf_verify_key_init,
-                self.taskprov_leader_token.clone(),
             ))
         }
 
@@ -381,13 +352,9 @@ mod test {
                     .gen_hpke_receiver_config_list(thread_rng().gen())
                     .expect("failed to generate HPKE receiver config"),
                 self.global_config,
-                self.leader_token,
-                self.collector_token.clone(),
                 self.collector_hpke_receiver_config.config.clone(),
                 &self.leader_registry,
                 self.taskprov_vdaf_verify_key_init,
-                self.taskprov_leader_token,
-                self.taskprov_collector_token.clone(),
                 Arc::clone(&helper),
             ));
 
@@ -395,8 +362,6 @@ mod test {
                 now: self.now,
                 leader,
                 helper,
-                collector_token: self.collector_token,
-                taskprov_collector_token: self.taskprov_collector_token,
                 time_interval_task_id: self.time_interval_task_id,
                 fixed_size_task_id: self.fixed_size_task_id,
                 expired_task_id: self.expired_task_id,
@@ -412,8 +377,6 @@ mod test {
         now: Time,
         leader: Arc<InMemoryAggregator>,
         helper: Arc<InMemoryAggregator>,
-        collector_token: BearerToken,
-        taskprov_collector_token: BearerToken,
         time_interval_task_id: TaskId,
         fixed_size_task_id: TaskId,
         expired_task_id: TaskId,
@@ -434,17 +397,19 @@ mod test {
             &self,
             report: Report,
             task_id: &TaskId,
-        ) -> DapRequest<BearerToken> {
+        ) -> 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: Some(*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,
             }
         }
 
@@ -452,7 +417,7 @@ mod test {
             &self,
             query: Query,
             task_id: &TaskId,
-        ) -> DapRequest<BearerToken> {
+        ) -> DapRequest<CollectionReq> {
             self.gen_test_coll_job_req_for_collection(query, DapAggregationParam::Empty, task_id)
                 .await
         }
@@ -462,13 +427,13 @@ mod test {
             query: Query,
             agg_param: DapAggregationParam,
             task_id: &TaskId,
-        ) -> DapRequest<BearerToken> {
+        ) -> DapRequest<CollectionReq> {
             let task_config = self.leader.unchecked_get_task_config(task_id).await;
 
-            self.collector_authorized_req(
+            Self::collector_req(
                 task_id,
                 &task_config,
-                DapMediaType::CollectReq,
+                DapMediaType::CollectionReq,
                 CollectionReq {
                     query,
                     agg_param: agg_param.get_encoded().unwrap(),
@@ -481,7 +446,7 @@ mod test {
             task_id: &TaskId,
             agg_param: DapAggregationParam,
             reports: Vec<Report>,
-        ) -> (DapAggregationJobState, DapRequest<BearerToken>) {
+        ) -> (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 {
@@ -508,40 +473,16 @@ mod test {
 
             (
                 leader_state,
-                self.leader_authorized_req(
+                Self::leader_req(
                     task_id,
                     &task_config,
                     Some(&agg_job_id),
                     DapMediaType::AggregationJobInitReq,
                     agg_job_init_req,
-                )
-                .await,
+                ),
             )
         }
 
-        pub async fn gen_test_agg_share_req(
-            &self,
-            report_count: u64,
-            checksum: [u8; 32],
-        ) -> DapRequest<BearerToken> {
-            let task_id = &self.time_interval_task_id;
-            let task_config = self.leader.unchecked_get_task_config(task_id).await;
-
-            self.leader_authorized_req(
-                task_id,
-                &task_config,
-                None,
-                DapMediaType::AggregateShareReq,
-                AggregateShareReq {
-                    batch_sel: BatchSelector::default(),
-                    agg_param: Vec::default(),
-                    report_count,
-                    checksum,
-                },
-            )
-            .await
-        }
-
         pub async fn gen_test_report(&self, task_id: &TaskId) -> Report {
             // Construct report. We expect the VDAF to be Prio3Count so that we know what type of
             // measurement to generate. However, we could extend the code to support more VDAFs.
@@ -585,57 +526,45 @@ mod test {
                 .unwrap()
         }
 
-        pub async fn leader_authorized_req<M: ParameterizedEncode<DapVersion>>(
-            &self,
+        pub fn leader_req<P>(
             task_id: &TaskId,
             task_config: &DapTaskConfig,
             agg_job_id: Option<&AggregationJobId>,
             media_type: DapMediaType,
-            msg: M,
-        ) -> DapRequest<BearerToken> {
-            let payload = msg.get_encoded_with_param(&task_config.version).unwrap();
-            let sender_auth = Some(
-                self.leader
-                    .authorize(task_id, task_config, &media_type, &payload)
-                    .await
-                    .unwrap(),
-            );
+            payload: P,
+        ) -> DapRequest<P> {
             DapRequest {
-                version: task_config.version,
-                media_type: Some(media_type),
-                task_id: Some(*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,
-                sender_auth,
-                ..Default::default()
             }
         }
 
-        pub fn collector_authorized_req<M: ParameterizedEncode<DapVersion>>(
-            &self,
+        pub fn collector_req<C>(
             task_id: &TaskId,
             task_config: &DapTaskConfig,
             media_type: DapMediaType,
-            msg: M,
-        ) -> DapRequest<BearerToken> {
+            payload: C,
+        ) -> DapRequest<C> {
             let mut rng = thread_rng();
             let coll_job_id = CollectionJobId(rng.gen());
-            let sender_auth = if task_config.method_is_taskprov() {
-                Some(self.taskprov_collector_token.clone())
-            } else {
-                Some(self.collector_token.clone())
-            };
 
             DapRequest {
-                version: task_config.version,
-                media_type: Some(media_type),
-                task_id: Some(*task_id),
-                resource: DapResource::CollectionJob(coll_job_id),
-                payload: msg.get_encoded_with_param(&task_config.version).unwrap(),
-                sender_auth,
-                ..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,
             }
         }
     }
@@ -649,23 +578,21 @@ mod test {
         let agg_job_id = AggregationJobId(rng.gen());
 
         // Helper expects "time_interval" query, but Leader indicates "fixed_size".
-        let req = t
-            .leader_authorized_req(
-                task_id,
-                &task_config,
-                Some(&agg_job_id),
-                DapMediaType::AggregationJobInitReq,
-                AggregationJobInitReq {
-                    agg_param: Vec::default(),
-                    part_batch_sel: PartialBatchSelector::FixedSizeByBatchId {
-                        batch_id: BatchId(rng.gen()),
-                    },
-                    prep_inits: Vec::default(),
+        let req = Test::leader_req(
+            task_id,
+            &task_config,
+            Some(&agg_job_id),
+            DapMediaType::AggregationJobInitReq,
+            AggregationJobInitReq {
+                agg_param: Vec::default(),
+                part_batch_sel: PartialBatchSelector::FixedSizeByBatchId {
+                    batch_id: BatchId(rng.gen()),
                 },
-            )
-            .await;
+                prep_inits: Vec::default(),
+            },
+        );
         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 { .. })
@@ -706,51 +633,13 @@ mod test {
     //
     //    async_test_versions! { handle_agg_job_req_init_expired_task }
 
-    async fn handle_agg_job_init_req_unauthorized_request(version: DapVersion) {
-        let t = Test::new(version);
-        let report = t.gen_test_report(&t.time_interval_task_id).await;
-        let (_, mut req) = t
-            .gen_test_agg_job_init_req(
-                &t.time_interval_task_id,
-                DapAggregationParam::Empty,
-                vec![report],
-            )
-            .await;
-        req.sender_auth = None;
-
-        // Expect failure due to missing bearer token.
-        assert_matches!(
-            helper::handle_agg_job_req(&*t.helper, &req, Default::default()).await,
-            Err(DapError::Abort(DapAbort::UnauthorizedRequest { .. }))
-        );
-
-        // Expect failure due to incorrect bearer token.
-        req.sender_auth = Some(BearerToken::from("incorrect auth token!".to_string()));
-        assert_matches!(
-            helper::handle_agg_job_req(&*t.helper, &req, Default::default()).await,
-            Err(DapError::Abort(DapAbort::UnauthorizedRequest { .. }))
-        );
-
-        assert_eq!(t.helper.audit_log.invocations(), 0);
-    }
-
-    async_test_versions! { handle_agg_job_init_req_unauthorized_request }
-
     async fn handle_hpke_config_req_unrecognized_task(version: DapVersion) {
         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: Some(task_id),
-            resource: DapResource::Undefined,
-            ..Default::default()
-        };
 
         assert_eq!(
-            aggregator::handle_hpke_config_req(&*t.leader, &req, Some(task_id))
+            aggregator::handle_hpke_config_req(&*t.leader, version, Some(task_id))
                 .await
                 .unwrap_err(),
             DapError::Abort(DapAbort::UnrecognizedTask { task_id })
@@ -761,47 +650,18 @@ 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: Some(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, None).await,
+            aggregator::handle_hpke_config_req(&*t.leader, version, None).await,
             Err(DapError::Abort(DapAbort::MissingTaskId))
         );
     }
 
     async_test_versions! { handle_hpke_config_req_missing_task_id }
 
-    async fn handle_agg_share_req_unauthorized_request(version: DapVersion) {
-        let t = Test::new(version);
-        let mut req = t.gen_test_agg_share_req(0, [0; 32]).await;
-        req.sender_auth = None;
-
-        // Expect failure due to missing bearer token.
-        assert_matches!(
-            helper::handle_agg_share_req(&*t.helper, &req).await,
-            Err(DapError::Abort(DapAbort::UnauthorizedRequest { .. }))
-        );
-
-        // Expect failure due to incorrect bearer token.
-        req.sender_auth = Some(BearerToken::from("incorrect auth token!".to_string()));
-        assert_matches!(
-            helper::handle_agg_share_req(&*t.helper, &req).await,
-            Err(DapError::Abort(DapAbort::UnauthorizedRequest { .. }))
-        );
-    }
-
-    async_test_versions! { handle_agg_share_req_unauthorized_request }
-
     // Test that the Helper handles the batch selector sent from the Leader properly.
     async fn handle_agg_share_req_invalid_batch_sel(version: DapVersion) {
         let mut rng = thread_rng();
@@ -812,24 +672,22 @@ mod test {
             .leader
             .unchecked_get_task_config(&t.time_interval_task_id)
             .await;
-        let req = t
-            .leader_authorized_req(
-                &t.time_interval_task_id,
-                &task_config,
-                None,
-                DapMediaType::AggregateShareReq,
-                AggregateShareReq {
-                    batch_sel: BatchSelector::FixedSizeByBatchId {
-                        batch_id: BatchId(rng.gen()),
-                    },
-                    agg_param: Vec::default(),
-                    report_count: 0,
-                    checksum: [0; 32],
+        let req = Test::leader_req(
+            &t.time_interval_task_id,
+            &task_config,
+            None,
+            DapMediaType::AggregateShareReq,
+            AggregateShareReq {
+                batch_sel: BatchSelector::FixedSizeByBatchId {
+                    batch_id: BatchId(rng.gen()),
                 },
-            )
-            .await;
+                agg_param: Vec::default(),
+                report_count: 0,
+                checksum: [0; 32],
+            },
+        );
         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 { .. })
@@ -840,24 +698,22 @@ mod test {
             .leader
             .unchecked_get_task_config(&t.fixed_size_task_id)
             .await;
-        let req = t
-            .leader_authorized_req(
-                &t.fixed_size_task_id,
-                &task_config,
-                None,
-                DapMediaType::AggregateShareReq,
-                AggregateShareReq {
-                    batch_sel: BatchSelector::FixedSizeByBatchId {
-                        batch_id: BatchId(rng.gen()), // Unrecognized batch ID
-                    },
-                    agg_param: Vec::default(),
-                    report_count: 0,
-                    checksum: [0; 32],
+        let req = Test::leader_req(
+            &t.fixed_size_task_id,
+            &task_config,
+            None,
+            DapMediaType::AggregateShareReq,
+            AggregateShareReq {
+                batch_sel: BatchSelector::FixedSizeByBatchId {
+                    batch_id: BatchId(rng.gen()), // Unrecognized batch ID
                 },
-            )
-            .await;
+                agg_param: Vec::default(),
+                report_count: 0,
+                checksum: [0; 32],
+            },
+        );
         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 { .. })
@@ -866,42 +722,6 @@ mod test {
 
     async_test_versions! { handle_agg_share_req_invalid_batch_sel }
 
-    async fn handle_coll_job_req_unauthorized_request(version: DapVersion) {
-        let mut rng = thread_rng();
-        let t = Test::new(version);
-        let task_id = &t.time_interval_task_id;
-        let task_config = t.leader.unchecked_get_task_config(task_id).await;
-        let collect_job_id = CollectionJobId(rng.gen());
-        let mut req = DapRequest {
-            version: task_config.version,
-            media_type: Some(DapMediaType::CollectReq),
-            task_id: Some(*task_id),
-            resource: DapResource::CollectionJob(collect_job_id),
-            payload: CollectionReq {
-                query: Query::default(),
-                agg_param: Vec::default(),
-            }
-            .get_encoded_with_param(&task_config.version)
-            .unwrap(),
-            ..Default::default() // Unauthorized request.
-        };
-
-        // Expect failure due to missing bearer token.
-        assert_matches!(
-            leader::handle_coll_job_req(&*t.leader, &req).await,
-            Err(DapError::Abort(DapAbort::UnauthorizedRequest { .. }))
-        );
-
-        // Expect failure due to incorrect bearer token.
-        req.sender_auth = Some(BearerToken::from("incorrect auth token!".to_string()));
-        assert_matches!(
-            leader::handle_coll_job_req(&*t.leader, &req).await,
-            Err(DapError::Abort(DapAbort::UnauthorizedRequest { .. }))
-        );
-    }
-
-    async_test_versions! { handle_coll_job_req_unauthorized_request }
-
     async fn handle_agg_job_req_failure_hpke_decrypt_error(version: DapVersion) {
         let t = Test::new(version);
         let task_id = &t.time_interval_task_id;
@@ -914,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,
@@ -942,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,
@@ -963,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;
@@ -1010,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.
@@ -1055,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: Some(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])
             }))
@@ -1084,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: Some(*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 {
@@ -1112,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);
@@ -1162,10 +984,10 @@ mod test {
         let task_config = t.leader.unchecked_get_task_config(task_id).await;
 
         // Collector: Create a CollectReq.
-        let req = t.collector_authorized_req(
+        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(),
@@ -1248,10 +1070,10 @@ mod test {
         let task_config = t.leader.unchecked_get_task_config(task_id).await;
 
         // Collector: Create a CollectReq with a very large batch interval.
-        let req = t.collector_authorized_req(
+        let req = Test::collector_req(
             task_id,
             &task_config,
-            DapMediaType::CollectReq,
+            DapMediaType::CollectionReq,
             CollectionReq {
                 query: Query::TimeInterval {
                     batch_interval: Interval {
@@ -1273,10 +1095,10 @@ mod test {
         assert_matches!(err, DapError::Abort(DapAbort::BadRequest(s)) => assert_eq!(s, "batch interval too large".to_string()));
 
         // Collector: Create a CollectReq with a batch interval in the past.
-        let req = t.collector_authorized_req(
+        let req = Test::collector_req(
             task_id,
             &task_config,
-            DapMediaType::CollectReq,
+            DapMediaType::CollectionReq,
             CollectionReq {
                 query: Query::TimeInterval {
                     batch_interval: Interval {
@@ -1299,10 +1121,10 @@ mod test {
         assert_matches!(err, DapError::Abort(DapAbort::BadRequest(s)) => assert_eq!(s, "batch interval too far into past".to_string()));
 
         // Collector: Create a CollectReq with a batch interval in the future.
-        let req = t.collector_authorized_req(
+        let req = Test::collector_req(
             task_id,
             &task_config,
-            DapMediaType::CollectReq,
+            DapMediaType::CollectionReq,
             CollectionReq {
                 query: Query::TimeInterval {
                     batch_interval: Interval {
@@ -1333,10 +1155,10 @@ mod test {
         let task_config = t.leader.unchecked_get_task_config(task_id).await;
 
         // Collector: Create a CollectReq with a very large batch interval.
-        let req = t.collector_authorized_req(
+        let req = Test::collector_req(
             task_id,
             &task_config,
-            DapMediaType::CollectReq,
+            DapMediaType::CollectionReq,
             CollectionReq {
                 query: Query::TimeInterval {
                     batch_interval: Interval {
@@ -1362,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;
@@ -1419,10 +1241,10 @@ mod test {
             query: task_config.query_for_current_batch_window(t.now),
             agg_param: Vec::default(),
         };
-        let req = t.collector_authorized_req(
+        let req = Test::collector_req(
             task_id,
             &task_config,
-            DapMediaType::CollectReq,
+            DapMediaType::CollectionReq,
             collector_collect_req.clone(),
         );
 
@@ -1472,10 +1294,10 @@ mod test {
             .leader
             .unchecked_get_task_config(&t.time_interval_task_id)
             .await;
-        let req = t.collector_authorized_req(
+        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()),
@@ -1495,10 +1317,10 @@ mod test {
             .leader
             .unchecked_get_task_config(&t.fixed_size_task_id)
             .await;
-        let req = t.collector_authorized_req(
+        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
@@ -1523,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");
     }
@@ -1540,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();
 
@@ -1576,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();
 
@@ -1657,15 +1479,16 @@ mod test {
                 .unwrap();
 
             let req = DapRequest {
-                version,
-                media_type: Some(DapMediaType::Report),
-                task_id: Some(task_id),
-                resource: DapResource::Undefined,
-                payload: report.get_encoded_with_param(&version).unwrap(),
-                taskprov: Some(taskprov_advertisement.clone()),
-                ..Default::default()
+                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.
@@ -1780,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();
 
@@ -1795,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();
 
@@ -1846,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 d1eac2a..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::{
@@ -106,8 +110,8 @@ 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<S>(
-    req: &'_ DapRequest<S>,
+pub(crate) fn resolve_advertised_task_config(
+    req: &DapRequestMeta,
     verify_key_init: &[u8; 32],
     collector_hpke_config: &HpkeConfig,
     task_id: &TaskId,
@@ -126,8 +130,8 @@ pub(crate) fn resolve_advertised_task_config<S>(
 }
 
 /// Check for a taskprov extension in the report, and return it if found.
-fn get_taskprov_task_config<S>(
-    req: &'_ DapRequest<S>,
+fn get_taskprov_task_config(
+    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,14 +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: Some(task_id),
-                resource: DapResource::Undefined, // ignored by test
-                payload: Vec::default(),          // ignored by test
-                sender_auth: None,                // ignored by test
+                task_id,
                 taskprov: Some(taskprov_task_config_base64url),
+                media_type: None,                 // ignored by test
+                resource: DapResource::Undefined, // ignored by test
             };
 
             (req, task_id)
@@ -677,14 +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: Some(task_id),
-                resource: DapResource::Undefined, // ignored by test
-                payload: Vec::default(),          // ignored by test
-                sender_auth: None,                // ignored by test
+                task_id,
                 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 2e340c9..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,
@@ -1087,6 +1087,7 @@ async fn fixed_size() {
 }
 
 async fn leader_collect_taskprov_ok(version: DapVersion) {
+    const DAP_TASKPROV_COLLECTOR_TOKEN: &str = "I-am-the-collector";
     let t = TestRunner::default_with_version(version).await;
     let batch_interval = t.batch_interval();
 
@@ -1149,7 +1150,7 @@ async fn leader_collect_taskprov_ok(version: DapVersion) {
     let collect_uri = t
         .leader_post_collect_using_token(
             client,
-            "I-am-the-collector", // DAP_TASKPROV_COLLECTOR_AUTH
+            DAP_TASKPROV_COLLECTOR_TOKEN,
             Some(&taskprov_advertisement),
             Some(&task_id),
             collect_req.get_encoded_with_param(&t.version).unwrap(),
@@ -1159,8 +1160,20 @@ async fn leader_collect_taskprov_ok(version: DapVersion) {
     println!("collect_uri: {collect_uri}");
 
     // Poll the collect URI before the CollectResp is ready.
-    let resp = t.poll_collection_url(client, &collect_uri).await.unwrap();
-    assert_eq!(resp.status(), 202, "response: {resp:?}");
+    let resp = t
+        .poll_collection_url_using_token(client, &collect_uri, DAP_TASKPROV_COLLECTOR_TOKEN)
+        .await
+        .unwrap();
+    #[allow(clippy::format_in_format_args)]
+    {
+        assert_eq!(
+            resp.status(),
+            202,
+            "response: {} {}",
+            format!("{resp:?}"),
+            resp.text().await.unwrap()
+        );
+    }
 
     // The reports are aggregated in the background.
     let agg_telem = t.internal_process(client).await.unwrap();
@@ -1178,7 +1191,10 @@ async fn leader_collect_taskprov_ok(version: DapVersion) {
     );
 
     // Poll the collect URI.
-    let resp = t.poll_collection_url(client, &collect_uri).await.unwrap();
+    let resp = t
+        .poll_collection_url_using_token(client, &collect_uri, DAP_TASKPROV_COLLECTOR_TOKEN)
+        .await
+        .unwrap();
     assert_eq!(resp.status(), 200);
 
     let collection =
@@ -1203,7 +1219,10 @@ async fn leader_collect_taskprov_ok(version: DapVersion) {
 
     // Poll the collect URI once more. Expect the response to be the same as the first, per HTTP
     // GET semantics.
-    let resp = t.poll_collection_url(client, &collect_uri).await.unwrap();
+    let resp = t
+        .poll_collection_url_using_token(client, &collect_uri, DAP_TASKPROV_COLLECTOR_TOKEN)
+        .await
+        .unwrap();
     assert_eq!(resp.status(), 200);
     assert_eq!(
         resp.bytes().await.unwrap(),
diff --git a/crates/daphne-server/tests/e2e/test_runner.rs b/crates/daphne-server/tests/e2e/test_runner.rs
index cb6c476..cc27207 100644
--- a/crates/daphne-server/tests/e2e/test_runner.rs
+++ b/crates/daphne-server/tests/e2e/test_runner.rs
@@ -72,6 +72,7 @@ impl TestRunner {
     }
 
     async fn with(version: DapVersion, query_config: &DapQueryConfig) -> Self {
+        println!("\n############ starting test prep ############");
         let mut rng = thread_rng();
         let now = SystemTime::now()
             .duration_since(SystemTime::UNIX_EPOCH)
@@ -485,6 +486,10 @@ impl TestRunner {
                 .context("no string for version")?
                 .parse()?,
         );
+        headers.insert(
+            reqwest::header::HeaderName::from_static(http_headers::DAP_AUTH_TOKEN),
+            reqwest::header::HeaderValue::from_str(&self.collector_bearer_token)?,
+        );
         if let Some(taskprov_advertisement) = taskprov {
             headers.insert(
                 reqwest::header::HeaderName::from_static(http_headers::DAP_TASKPROV),
@@ -585,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")?,
             )?,
@@ -643,8 +648,8 @@ impl TestRunner {
         anyhow::ensure!(
             resp.status() == 200,
             "unexpected response status. Expected {} got {}: Body is {:?}",
-            resp.status(),
             reqwest::StatusCode::OK,
+            resp.status(),
             resp.text().await?,
         );
         Ok(resp.json().await?)
@@ -671,7 +676,9 @@ impl TestRunner {
             .context("request failed")?;
         anyhow::ensure!(
             resp.status() == 200,
-            "request to {url} failed: response: {resp:?}"
+            "request to {url} failed: response: {} {}",
+            format!("{resp:?}"), // text() moves so we have to format here
+            resp.text().await.unwrap_or("<failed to get body>".into()),
         );
         let t = resp.text().await.context("failed to extract text")?;
         // This is needed so we can have tests that call this expecting nothing and have it work
@@ -756,17 +763,31 @@ impl TestRunner {
         &self,
         client: &reqwest::Client,
         url: &Url,
+    ) -> anyhow::Result<reqwest::Response> {
+        self.poll_collection_url_using_token(client, url, &self.collector_bearer_token)
+            .await
+    }
+
+    pub async fn poll_collection_url_using_token(
+        &self,
+        client: &reqwest::Client,
+        url: &Url,
+        token: &str,
     ) -> anyhow::Result<reqwest::Response> {
         let builder = client.post(url.as_str());
         let mut headers = reqwest::header::HeaderMap::new();
         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")?,
             )?,
         );
+        headers.insert(
+            reqwest::header::HeaderName::from_static(http_headers::DAP_AUTH_TOKEN),
+            reqwest::header::HeaderValue::from_str(token)?,
+        );
         Ok(builder.headers(headers).send().await?)
     }
 }
diff --git a/crates/daphne/src/testing/mod.rs b/crates/daphne/src/testing/mod.rs
index e558d21..f2f4ff4 100644
--- a/crates/daphne/src/testing/mod.rs
+++ b/crates/daphne/src/testing/mod.rs
@@ -8,7 +8,6 @@ pub mod report_generator;
 
 use crate::{
     audit_log::AuditLog,
-    auth::{BearerToken, BearerTokenProvider},
     constants::DapMediaType,
     fatal_error,
     hpke::{HpkeConfig, HpkeDecrypter, HpkeKemId, HpkeProvider, HpkeReceiverConfig},
@@ -23,17 +22,18 @@ use crate::{
         aggregator::MergeAggShareError,
         helper,
         leader::{in_memory_leader::InMemoryLeaderState, WorkItem},
-        DapAggregator, DapAuthorizedSender, DapHelper, DapLeader, DapReportInitializer,
+        DapAggregator, DapHelper, DapLeader, DapReportInitializer,
     },
     taskprov,
     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::{
@@ -577,8 +577,6 @@ pub struct InMemoryAggregator {
     pub(crate) global_config: DapGlobalConfig,
     tasks: Mutex<HashMap<TaskId, DapTaskConfig>>,
     pub hpke_receiver_config_list: Box<[HpkeReceiverConfig]>,
-    leader_token: BearerToken,
-    collector_token: Option<BearerToken>, // Not set by Helper
     collector_hpke_config: HpkeConfig,
 
     // aggregation state
@@ -591,8 +589,6 @@ pub struct InMemoryAggregator {
 
     // taskprov
     taskprov_vdaf_verify_key_init: [u8; 32],
-    taskprov_leader_token: BearerToken,
-    taskprov_collector_token: Option<BearerToken>, // Not set by Helper
 
     // Leader: Reference to peer. Used to simulate HTTP requests from Leader to Helper, i.e.,
     // implement `DapLeader::send_http_post()` for `InMemoryAggregator`. Not set by the Helper.
@@ -605,29 +601,21 @@ impl DeepSizeOf for InMemoryAggregator {
             global_config,
             tasks,
             hpke_receiver_config_list,
-            leader_token,
-            collector_token,
             leader_state_store,
             agg_store,
             collector_hpke_config,
             metrics: _,
             audit_log: _,
             taskprov_vdaf_verify_key_init,
-            taskprov_leader_token,
-            taskprov_collector_token,
             peer,
         } = self;
         global_config.deep_size_of_children(context)
             + tasks.deep_size_of_children(context)
             + hpke_receiver_config_list.deep_size_of_children(context)
-            + leader_token.deep_size_of_children(context)
-            + collector_token.deep_size_of_children(context)
             + leader_state_store.deep_size_of_children(context)
             + agg_store.deep_size_of_children(context)
             + collector_hpke_config.deep_size_of_children(context)
             + taskprov_vdaf_verify_key_init.deep_size_of_children(context)
-            + taskprov_leader_token.deep_size_of_children(context)
-            + taskprov_collector_token.deep_size_of_children(context)
             + peer.deep_size_of_children(context)
     }
 }
@@ -638,26 +626,20 @@ impl InMemoryAggregator {
         tasks: impl IntoIterator<Item = (TaskId, DapTaskConfig)>,
         hpke_receiver_config_list: impl IntoIterator<Item = HpkeReceiverConfig>,
         global_config: DapGlobalConfig,
-        leader_token: BearerToken,
         collector_hpke_config: HpkeConfig,
         registry: &prometheus::Registry,
         taskprov_vdaf_verify_key_init: [u8; 32],
-        taskprov_leader_token: BearerToken,
     ) -> Self {
         Self {
             global_config,
             tasks: Mutex::new(tasks.into_iter().collect()),
             hpke_receiver_config_list: hpke_receiver_config_list.into_iter().collect(),
-            leader_token,
-            collector_token: None,
             leader_state_store: Default::default(),
             agg_store: Default::default(),
             collector_hpke_config,
             metrics: DaphnePromMetrics::register(registry).unwrap(),
             audit_log: MockAuditLog::default(),
             taskprov_vdaf_verify_key_init,
-            taskprov_leader_token,
-            taskprov_collector_token: None,
             peer: None,
         }
     }
@@ -667,29 +649,21 @@ impl InMemoryAggregator {
         tasks: impl IntoIterator<Item = (TaskId, DapTaskConfig)>,
         hpke_receiver_config_list: impl IntoIterator<Item = HpkeReceiverConfig>,
         global_config: DapGlobalConfig,
-        leader_token: BearerToken,
-        collector_token: impl Into<Option<BearerToken>>,
         collector_hpke_config: HpkeConfig,
         registry: &prometheus::Registry,
         taskprov_vdaf_verify_key_init: [u8; 32],
-        taskprov_leader_token: BearerToken,
-        taskprov_collector_token: impl Into<Option<BearerToken>>,
         peer: Arc<Self>,
     ) -> Self {
         Self {
             global_config,
             tasks: Mutex::new(tasks.into_iter().collect()),
             hpke_receiver_config_list: hpke_receiver_config_list.into_iter().collect(),
-            leader_token,
-            collector_token: collector_token.into(),
             leader_state_store: Default::default(),
             agg_store: Default::default(),
             collector_hpke_config,
             metrics: DaphnePromMetrics::register(registry).unwrap(),
             audit_log: MockAuditLog::default(),
             taskprov_vdaf_verify_key_init,
-            taskprov_leader_token,
-            taskprov_collector_token: taskprov_collector_token.into(),
             peer: peer.into(),
         }
     }
@@ -717,39 +691,6 @@ impl InMemoryAggregator {
     }
 }
 
-#[async_trait]
-impl BearerTokenProvider for InMemoryAggregator {
-    type WrappedBearerToken<'a> = &'a BearerToken;
-
-    async fn get_leader_bearer_token_for<'s>(
-        &'s self,
-        _task_id: &'s TaskId,
-        task_config: &DapTaskConfig,
-    ) -> Result<Option<Self::WrappedBearerToken<'s>>, DapError> {
-        if task_config.method_is_taskprov() {
-            Ok(Some(&self.taskprov_leader_token))
-        } else {
-            Ok(Some(&self.leader_token))
-        }
-    }
-
-    async fn get_collector_bearer_token_for<'s>(
-        &'s self,
-        _task_id: &'s TaskId,
-        task_config: &DapTaskConfig,
-    ) -> Result<Option<Self::WrappedBearerToken<'s>>, DapError> {
-        if task_config.method_is_taskprov() {
-            Ok(Some(self.taskprov_collector_token.as_ref().expect(
-                "InMemoryAggregator not configured with taskprov collector token",
-            )))
-        } else {
-            Ok(Some(self.collector_token.as_ref().expect(
-                "InMemoryAggregator not configured with collector token",
-            )))
-        }
-    }
-}
-
 #[async_trait]
 impl HpkeProvider for InMemoryAggregator {
     type WrappedHpkeConfig<'a> = &'a HpkeConfig;
@@ -799,22 +740,6 @@ impl HpkeDecrypter for InMemoryAggregator {
     }
 }
 
-#[async_trait]
-impl DapAuthorizedSender<BearerToken> for InMemoryAggregator {
-    async fn authorize(
-        &self,
-        task_id: &TaskId,
-        task_config: &DapTaskConfig,
-        media_type: &DapMediaType,
-        _payload: &[u8],
-    ) -> Result<BearerToken, DapError> {
-        Ok(self
-            .authorize_with_bearer_token(task_id, task_config, media_type)
-            .await?
-            .clone())
-    }
-}
-
 #[async_trait]
 impl DapReportInitializer for InMemoryAggregator {
     fn valid_report_time_range(&self) -> Range<messages::Time> {
@@ -841,20 +766,12 @@ impl DapReportInitializer for InMemoryAggregator {
 }
 
 #[async_trait]
-impl DapAggregator<BearerToken> for InMemoryAggregator {
+impl DapAggregator for InMemoryAggregator {
     // The lifetimes on the traits ensure that we can return a reference to a task config stored by
     // the DapAggregator. (See DaphneWorkerConfig for an example.) For simplicity, InMemoryAggregator
     // clones the task config as needed.
     type WrappedDapTaskConfig<'a> = DapTaskConfig;
 
-    async fn unauthorized_reason(
-        &self,
-        task_config: &DapTaskConfig,
-        req: &DapRequest<BearerToken>,
-    ) -> Result<Option<String>, DapError> {
-        self.bearer_token_authorized(task_config, req).await
-    }
-
     async fn get_global_config(&self) -> Result<DapGlobalConfig, DapError> {
         Ok(self.global_config.clone())
     }
@@ -882,12 +799,11 @@ impl DapAggregator<BearerToken> for InMemoryAggregator {
 
     async fn taskprov_put(
         &self,
-        req: &DapRequest<BearerToken>,
+        req: &DapRequestMeta,
         task_config: DapTaskConfig,
     ) -> Result<(), DapError> {
-        let task_id = req.task_id().map_err(DapError::Abort)?;
         let mut tasks = self.tasks.lock().expect("tasks: lock failed");
-        tasks.deref_mut().insert(*task_id, task_config);
+        tasks.deref_mut().insert(req.task_id, task_config);
         Ok(())
     }
 
@@ -1063,10 +979,10 @@ impl DapAggregator<BearerToken> for InMemoryAggregator {
 }
 
 #[async_trait]
-impl DapHelper<BearerToken> for InMemoryAggregator {}
+impl DapHelper for InMemoryAggregator {}
 
 #[async_trait]
-impl DapLeader<BearerToken> for InMemoryAggregator {
+impl DapLeader for InMemoryAggregator {
     async fn put_report(&self, report: &Report, task_id: &TaskId) -> Result<(), DapError> {
         let task_config = self
             .get_task_config_for(task_id)
@@ -1152,22 +1068,25 @@ impl DapLeader<BearerToken> for InMemoryAggregator {
             .finish_collect_job(task_id, coll_job_id, collection)
     }
 
-    async fn send_http_post(
+    async fn send_http_post<M>(
         &self,
-        req: DapRequest<BearerToken>,
+        req: DapRequest<M>,
         _url: Url,
-    ) -> Result<DapResponse, DapError> {
+    ) -> 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")),
@@ -1175,15 +1094,14 @@ impl DapLeader<BearerToken> for InMemoryAggregator {
         }
     }
 
-    async fn send_http_put(
-        &self,
-        req: DapRequest<BearerToken>,
-        _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
@@ -1194,6 +1112,24 @@ impl DapLeader<BearerToken> 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))]