Repository: cloudflare/daphne
Pull request #152 (2022-10-25 19:49:19)

taskprov support


Patch:
diff --git a/daphne/src/auth.rs b/daphne/src/auth.rs
index bb67016..e785418 100644
--- a/daphne/src/auth.rs
+++ b/daphne/src/auth.rs
@@ -70,6 +70,12 @@ pub trait BearerTokenProvider<'a> {
         task_id: &'a Id,
     ) -> Result<Option<Self::WrappedBearerToken>, DapError>;
 
+    /// Returns true if the given bearer token matches the leader token configured for the "taskprov" extension.
+    fn is_taskprov_leader_bearer_token(&self, token: &BearerToken) -> bool;
+
+    /// Returns true if the given bearer token matches the collector token configured for the "taskprov" extension.
+    fn is_taskprov_collector_bearer_token(&self, token: &BearerToken) -> bool;
+
     /// 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(
@@ -118,6 +124,7 @@ pub trait BearerTokenProvider<'a> {
                 if let Some(expected) = self.get_leader_bearer_token_for(task_id).await? {
                     return Ok(got == expected.as_ref());
                 }
+                return Ok(self.is_taskprov_leader_bearer_token(got));
             }
         }
 
@@ -126,6 +133,7 @@ pub trait BearerTokenProvider<'a> {
                 if let Some(expected) = self.get_collector_bearer_token_for(task_id).await? {
                     return Ok(got == expected.as_ref());
                 }
+                return Ok(self.is_taskprov_collector_bearer_token(got));
             }
         }
 
diff --git a/daphne/src/hpke.rs b/daphne/src/hpke.rs
index 9a8f6df..701e4a6 100644
--- a/daphne/src/hpke.rs
+++ b/daphne/src/hpke.rs
@@ -255,6 +255,11 @@ impl HpkeReceiverConfig {
             }
         }
     }
+
+    /// Create a new HPKE receiver context given an HpkeConfig and a corresponding secret key.
+    pub fn new(config: HpkeConfig, secret_key: Vec<u8>) -> Self {
+        HpkeReceiverConfig { config, secret_key }
+    }
 }
 
 #[async_trait(?Send)]
diff --git a/daphne/src/lib.rs b/daphne/src/lib.rs
index 5f274a0..9c188b9 100644
--- a/daphne/src/lib.rs
+++ b/daphne/src/lib.rs
@@ -347,6 +347,9 @@ pub struct DapGlobalConfig {
     /// HPKE KEM types that are supported. Used when generating HPKE
     /// receiver config.
     pub supported_hpke_kems: Vec<HpkeKemId>,
+
+    /// Is the taskprov extension allowed?
+    pub allow_taskprov: bool,
 }
 
 impl DapGlobalConfig {
@@ -884,5 +887,8 @@ pub mod messages;
 pub mod roles;
 #[cfg(test)]
 mod roles_test;
+pub mod taskprov;
+#[cfg(test)]
+mod taskprov_test;
 pub mod testing;
 pub mod vdaf;
diff --git a/daphne/src/messages/mod.rs b/daphne/src/messages/mod.rs
index 3f52c4a..690f7df 100644
--- a/daphne/src/messages/mod.rs
+++ b/daphne/src/messages/mod.rs
@@ -9,6 +9,7 @@ use prio::codec::{
 };
 use serde::{Deserialize, Serialize};
 use std::{
+    collections::HashSet,
     convert::{TryFrom, TryInto},
     fmt::Debug,
     io::{Cursor, Read},
@@ -24,6 +25,9 @@ const AEAD_ID_AES128GCM: u16 = 0x0001;
 const QUERY_TYPE_TIME_INTERVAL: u8 = 0x01;
 const QUERY_TYPE_FIXED_SIZE: u8 = 0x02;
 
+// Known extension types.
+const EXTENSION_TASKPROV: u16 = 0xff00;
+
 /// The identifier for a DAP task.
 #[derive(Clone, Debug, Default, Deserialize, Hash, PartialEq, Eq, Serialize)]
 pub struct Id(#[serde(with = "hex")] pub [u8; 32]);
@@ -101,12 +105,27 @@ impl AsRef<[u8]> for ReportId {
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "snake_case")]
 pub enum Extension {
+    Taskprov { payload: Vec<u8> }, // Not a TaskConfig to make computing the expected task id more efficient
     Unhandled { typ: u16, payload: Vec<u8> },
 }
 
+impl Extension {
+    /// Return the type code associated with the extension
+    fn type_code(&self) -> u16 {
+        match self {
+            Self::Taskprov { .. } => EXTENSION_TASKPROV,
+            Self::Unhandled { typ, .. } => *typ,
+        }
+    }
+}
+
 impl Encode for Extension {
     fn encode(&self, bytes: &mut Vec<u8>) {
         match self {
+            Self::Taskprov { payload } => {
+                EXTENSION_TASKPROV.encode(bytes);
+                encode_u16_bytes(bytes, payload);
+            }
             Self::Unhandled { typ, payload } => {
                 typ.encode(bytes);
                 encode_u16_bytes(bytes, payload);
@@ -117,10 +136,12 @@ impl Encode for Extension {
 
 impl Decode for Extension {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
-        Ok(Self::Unhandled {
-            typ: u16::decode(bytes)?,
-            payload: decode_u16_bytes(bytes)?,
-        })
+        let typ = u16::decode(bytes)?;
+        let payload = decode_u16_bytes(bytes)?;
+        match typ {
+            EXTENSION_TASKPROV => Ok(Self::Taskprov { payload }),
+            _ => Ok(Self::Unhandled { typ, payload }),
+        }
     }
 }
 
@@ -143,11 +164,19 @@ impl Encode for ReportMetadata {
 
 impl Decode for ReportMetadata {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
-        Ok(Self {
+        let metadata = Self {
             id: ReportId::decode(bytes)?,
             time: Time::decode(bytes)?,
             extensions: decode_u16_items(&(), bytes)?,
-        })
+        };
+        // Check for duplicate extensions.
+        let mut seen: HashSet<u16> = HashSet::new();
+        for extension in &metadata.extensions {
+            if !seen.insert(extension.type_code()) {
+                return Err(CodecError::UnexpectedValue);
+            }
+        }
+        Ok(metadata)
     }
 }
 
diff --git a/daphne/src/messages/taskprov.rs b/daphne/src/messages/taskprov.rs
index 7d79c95..4dcb559 100644
--- a/daphne/src/messages/taskprov.rs
+++ b/daphne/src/messages/taskprov.rs
@@ -5,17 +5,14 @@
 //! defined in draft-wang-ppm-dap-taskprov-00.
 
 use crate::messages::{
-    decode_u16_bytes, encode_u16_bytes, Duration, Id, Time, QUERY_TYPE_FIXED_SIZE,
+    decode_u16_bytes, encode_u16_bytes, Duration, Time, QUERY_TYPE_FIXED_SIZE,
     QUERY_TYPE_TIME_INTERVAL,
 };
 use prio::codec::{
     decode_u16_items, decode_u24_items, decode_u8_items, encode_u16_items, encode_u24_items,
     encode_u8_items, CodecError, Decode, Encode,
 };
-use ring::{
-    hmac::HMAC_SHA256,
-    hmac::{sign, Key},
-};
+use ring::hkdf::KeyType;
 use serde::{Deserialize, Serialize};
 use std::io::Cursor;
 
@@ -29,6 +26,29 @@ const VDAF_TYPE_POPLAR1_AES128: u32 = 0x00001000; // The gap from the previous c
 const DP_MECHANISM_RESERVED: u8 = 0x00;
 const DP_MECHANISM_NONE: u8 = 0x01;
 
+/// A VDAF type.
+#[derive(Clone, Deserialize, Serialize, Debug, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub(crate) enum VdafType {
+    Prio3Aes128Count,
+    Prio3Aes128Sum,
+    Prio3Aes128Histogram,
+    Poplar1Aes128,
+    NotImplemented(u32),
+}
+
+impl KeyType for VdafType {
+    fn len(&self) -> usize {
+        match self {
+            VdafType::Prio3Aes128Count => 16,
+            VdafType::Prio3Aes128Sum => 16,
+            VdafType::Prio3Aes128Histogram => 16,
+            VdafType::Poplar1Aes128 => 16,
+            _ => panic!("tried to get key length for undefined VDAF"),
+        }
+    }
+}
+
 /// A VDAF type along with its type-specific data.
 #[derive(Clone, Deserialize, Serialize, Debug, PartialEq, Eq)]
 pub enum VdafTypeVar {
@@ -81,6 +101,18 @@ impl Decode for VdafTypeVar {
     }
 }
 
+impl From<VdafTypeVar> for VdafType {
+    fn from(var: VdafTypeVar) -> Self {
+        match var {
+            VdafTypeVar::Prio3Aes128Count => VdafType::Prio3Aes128Count,
+            VdafTypeVar::Prio3Aes128Histogram { .. } => VdafType::Prio3Aes128Histogram,
+            VdafTypeVar::Prio3Aes128Sum { .. } => VdafType::Prio3Aes128Sum,
+            VdafTypeVar::Poplar1Aes128 { .. } => VdafType::Poplar1Aes128,
+            VdafTypeVar::NotImplemented(x) => VdafType::NotImplemented(x),
+        }
+    }
+}
+
 /// A differential privacy mechanism.
 #[derive(Clone, Copy, Deserialize, Serialize, Debug, PartialEq, Eq)]
 pub enum DpMechanism {
@@ -255,24 +287,6 @@ pub struct TaskConfig {
     pub vdaf_config: VdafConfig,
 }
 
-impl TaskConfig {
-    pub fn compute_task_id(&self) -> Id {
-        // SHA-256 of "dap-taskprov-00"
-        let task_prov_salt: Vec<u8> = vec![
-            0x4d, 0x63, 0x1a, 0xeb, 0xa8, 0xdf, 0xe0, 0x1b, 0x34, 0x4c, 0x29, 0x2d, 0x17, 0xba,
-            0x34, 0x9a, 0x78, 0x97, 0xbf, 0x64, 0x88, 0x00, 0x55, 0x1c, 0x0d, 0x75, 0x32, 0xab,
-            0x61, 0x4b, 0xe2, 0x21,
-        ];
-        let key = Key::new(HMAC_SHA256, &task_prov_salt);
-        let encoded = self.get_encoded();
-        let digest = sign(&key, &encoded);
-        let mut b: [u8; 32] = [0; 32];
-        let d = digest.as_ref();
-        b[..32].copy_from_slice(&d[..32]);
-        Id(b)
-    }
-}
-
 impl Encode for TaskConfig {
     fn encode(&self, bytes: &mut Vec<u8>) {
         encode_u8_items(bytes, &(), &self.task_info);
diff --git a/daphne/src/roles.rs b/daphne/src/roles.rs
index fc0fba0..7133936 100644
--- a/daphne/src/roles.rs
+++ b/daphne/src/roles.rs
@@ -55,11 +55,21 @@ where
     fn get_global_config(&self) -> &DapGlobalConfig;
 
     /// Look up the DAP task configuration for the given task ID.
-    async fn get_task_config_for(
+    async fn get_task_config_considering_taskprov(
         &'srv self,
         task_id: Cow<'req, Id>,
+        report: Option<&ReportMetadata>,
     ) -> Result<Option<Self::WrappedDapTaskConfig>, DapError>;
 
+    /// Look up the DAP task configuration for the given task ID.
+    async fn get_task_config_for(
+        &'srv self,
+        task_id: Cow<'req, Id>,
+    ) -> Result<Option<Self::WrappedDapTaskConfig>, DapError> {
+        self.get_task_config_considering_taskprov(task_id, None)
+            .await
+    }
+
     /// Get the current time (number of seconds since the beginning of UNIX time).
     fn get_current_time(&self) -> Time;
 
@@ -233,7 +243,10 @@ where
 
         let report = Report::get_decoded(req.payload.as_ref())?;
         let task_config = self
-            .get_task_config_for(Cow::Borrowed(req.task_id()?))
+            .get_task_config_considering_taskprov(
+                Cow::Borrowed(req.task_id()?),
+                Some(&report.metadata),
+            )
             .await?
             .ok_or(DapAbort::UnrecognizedTask)?;
 
@@ -575,8 +588,37 @@ where
         match req.media_type {
             Some(MEDIA_TYPE_AGG_INIT_REQ) => {
                 let agg_init_req = AggregateInitializeReq::get_decoded(&req.payload)?;
+
+                let mut first_metadata: Option<&ReportMetadata> = None;
+
+                // If taskprov is allowed, ensure that either all of the shares have it or none of them
+                // do (section 6 of draft-wang-ppm-dap-taskprov-00).
+                if self.get_global_config().allow_taskprov {
+                    let task_id = req.task_id()?;
+                    let using_taskprov = agg_init_req
+                        .report_shares
+                        .iter()
+                        .filter(|share| share.metadata.is_taskprov(task_id))
+                        .count();
+
+                    if using_taskprov == agg_init_req.report_shares.len() {
+                        // All the extensions use taskprov and look ok, so compute first_metadata.
+                        // Note this will always be Some().
+                        first_metadata = agg_init_req
+                            .report_shares
+                            .first()
+                            .map(|report_share| &report_share.metadata);
+                    } else if using_taskprov != 0 {
+                        // It's not all taskprov or no taskprov, so it's an error.
+                        return Err(DapAbort::UnrecognizedMessage);
+                    }
+                }
+
                 let wrapped_task_config = self
-                    .get_task_config_for(Cow::Borrowed(req.task_id()?))
+                    .get_task_config_considering_taskprov(
+                        Cow::Borrowed(req.task_id()?),
+                        first_metadata,
+                    )
                     .await?
                     .ok_or(DapAbort::UnrecognizedTask)?;
                 let task_config = wrapped_task_config.as_ref();
diff --git a/daphne/src/taskprov.rs b/daphne/src/taskprov.rs
new file mode 100644
index 0000000..07a390a
--- /dev/null
+++ b/daphne/src/taskprov.rs
@@ -0,0 +1,193 @@
+// Copyright (c) 2022 Cloudflare, Inc. All rights reserved.
+// SPDX-License-Identifier: BSD-3-Clause
+
+use crate::{
+    messages::{
+        taskprov::{QueryConfigVar, TaskConfig, VdafType, VdafTypeVar},
+        Extension, HpkeConfig, Id, ReportMetadata,
+    },
+    vdaf::VdafVerifyKey,
+    DapAbort, DapError, DapQueryConfig, DapTaskConfig, DapVersion, Prio3Config, VdafConfig,
+};
+use prio::codec::Decode;
+use ring::{
+    hkdf::{Prk, Salt, HKDF_SHA256},
+    hmac::HMAC_SHA256,
+    hmac::{sign, Key},
+};
+use std::str;
+use url::Url;
+
+/// SHA-256 of "dap-taskprov-00"
+#[allow(dead_code)]
+pub(crate) const TASK_PROV_SALT: [u8; 32] = [
+    0x4d, 0x63, 0x1a, 0xeb, 0xa8, 0xdf, 0xe0, 0x1b, 0x34, 0x4c, 0x29, 0x2d, 0x17, 0xba, 0x34, 0x9a,
+    0x78, 0x97, 0xbf, 0x64, 0x88, 0x00, 0x55, 0x1c, 0x0d, 0x75, 0x32, 0xab, 0x61, 0x4b, 0xe2, 0x21,
+];
+
+/// Compute the task id of a serialized task config.
+pub fn compute_task_id(serialized: &[u8]) -> Id {
+    // This is an implementation of HKDF-Expand as the ring::hkdf API does not expose the result
+    // of the expand step.
+    let key = Key::new(HMAC_SHA256, &TASK_PROV_SALT);
+    let digest = sign(&key, serialized);
+    let mut b: [u8; 32] = [0; 32];
+    let d = digest.as_ref();
+    b[..32].copy_from_slice(&d[..32]);
+    Id(b)
+}
+
+// The documentation for ring::hkdf says computing the Salt is expensive, and we use the same PRK all the
+// time, so we have structured the code so it would be possible to generate the PRK once and then
+// compute a VDAF verify key many times from the PRK, but we do not actually exploit this currently.
+
+/// Extract pseudorandom key from the pre-shared secret used for the "taskprov" extension.
+pub(crate) fn extract_prk_from_verify_key_init(verify_key_init: &[u8]) -> Prk {
+    // The documentation says computing the Salt is expensive, and we use the same PRK all the
+    // time, so we compute it once.
+    Salt::new(HKDF_SHA256, &TASK_PROV_SALT).extract(verify_key_init)
+}
+
+/// Expand a pseudorandom key into the VDAF verification key for a given task.
+pub(crate) fn expand_prk_into_verify_key(
+    prk: &Prk,
+    task_id: &Id,
+    vdaf_type: VdafType,
+) -> VdafVerifyKey {
+    let info = [task_id.as_ref()];
+    match &vdaf_type {
+        VdafType::Prio3Aes128Count | VdafType::Prio3Aes128Sum | VdafType::Prio3Aes128Histogram => {
+            // This expand(), and the associated fill() below can only fail if the length is wrong,
+            // and it won't be, so we unwrap().
+            let okm = prk.expand(&info, vdaf_type.clone()).unwrap();
+            let mut bytes = [0u8; 16];
+            okm.fill(&mut bytes[..]).unwrap();
+            VdafVerifyKey::Prio3(bytes)
+        }
+        _ => panic!("Unknown VDAF type"),
+    }
+}
+
+/// Compute the VDAF verify key for task_id and the specified VDAF type using the
+/// pre-shared secret verify_key_init.
+///
+/// This is a convenience function to call compute_vdaf_verify_prk() and
+/// compute_vdaf_verify_key_from_prk(). Callers reusing the same PRK frequently
+/// should consider computing the prk once and then calling compute_vdaf_verify_key_from_prk()
+/// directly.
+#[allow(dead_code)]
+pub(crate) fn compute_vdaf_verify_key(
+    verify_key_init: &[u8],
+    task_id: &Id,
+    vdaf_type: VdafType,
+) -> VdafVerifyKey {
+    expand_prk_into_verify_key(
+        &extract_prk_from_verify_key_init(verify_key_init),
+        task_id,
+        vdaf_type,
+    )
+}
+
+pub fn bad_request(detail: &str) -> DapError {
+    DapError::Abort(DapAbort::BadRequest(detail.to_string()))
+}
+
+/// Check for a taskprov extension in the report, and return it if found.
+pub fn get_taskprov_task_config(
+    task_id: &Id,
+    metadata: &ReportMetadata,
+) -> Result<Option<TaskConfig>, DapError> {
+    let taskprovs: Vec<&Extension> = metadata
+        .extensions
+        .iter()
+        .filter(|x| matches!(x, Extension::Taskprov { .. }))
+        .collect();
+    match taskprovs.len() {
+        0 => Ok(None),
+        1 => match &taskprovs[0] {
+            Extension::Taskprov { payload } => {
+                if compute_task_id(&payload[..]) != *task_id {
+                    // Return unrecognizedTask following section 5.1 of the taskprov draft.
+                    return Err(DapError::Abort(DapAbort::UnrecognizedTask));
+                }
+                // Return unrecognizedMessage if parsing fails following section 5.1 of the taskprov draft.
+                let task_config = TaskConfig::get_decoded(payload)
+                    .map_err(|_| DapError::Abort(DapAbort::UnrecognizedMessage))?;
+                Ok(Some(task_config))
+            }
+            _ => panic!("cannot happen"),
+        },
+        _ => {
+            // The decoder already returns an error if an extension of a give type occurs more than once.
+            panic!("should not happen")
+        }
+    }
+}
+
+fn url_from_bytes(bytes: &[u8]) -> Result<Url, DapError> {
+    let s = str::from_utf8(bytes).map_err(|_| bad_request("bad URL UTF8"))?;
+    Url::parse(s).map_err(|_| bad_request("bad URL syntax"))
+}
+
+impl From<QueryConfigVar> for DapQueryConfig {
+    fn from(var: QueryConfigVar) -> Self {
+        match var {
+            QueryConfigVar::FixedSize { max_batch_size } => DapQueryConfig::FixedSize {
+                max_batch_size: max_batch_size.into(),
+            },
+            QueryConfigVar::TimeInterval => DapQueryConfig::TimeInterval,
+        }
+    }
+}
+
+impl From<VdafTypeVar> for VdafConfig {
+    fn from(var: VdafTypeVar) -> Self {
+        match var {
+            VdafTypeVar::Prio3Aes128Count => VdafConfig::Prio3(Prio3Config::Count),
+            VdafTypeVar::Prio3Aes128Histogram { buckets } => {
+                VdafConfig::Prio3(Prio3Config::Histogram { buckets })
+            }
+            VdafTypeVar::Prio3Aes128Sum { bit_length } => VdafConfig::Prio3(Prio3Config::Sum {
+                bits: bit_length.into(),
+            }),
+            _ => panic!("poplar1 is not implemented"),
+        }
+    }
+}
+
+impl DapTaskConfig {
+    pub fn try_from_taskprov(
+        version: DapVersion,
+        task_id: &Id,
+        task_config: TaskConfig,
+        vdaf_verify_key_init: &[u8],
+        collector_hpke_config: &HpkeConfig,
+    ) -> Result<DapTaskConfig, DapError> {
+        if task_config.aggregator_endpoints.len() != 2 {
+            return Err(bad_request("number of aggregator endpoints is not 2"));
+        }
+        let vdaf_type = VdafType::from(task_config.vdaf_config.var.clone());
+        Ok(DapTaskConfig {
+            version,
+            leader_url: url_from_bytes(&task_config.aggregator_endpoints[0].bytes)?,
+            helper_url: url_from_bytes(&task_config.aggregator_endpoints[1].bytes)?,
+            time_precision: task_config.query_config.time_precision,
+            expiration: task_config.task_expiration,
+            min_batch_size: task_config.query_config.min_batch_size.into(),
+            query: DapQueryConfig::from(task_config.query_config.var),
+            vdaf: VdafConfig::from(task_config.vdaf_config.var),
+            vdaf_verify_key: compute_vdaf_verify_key(vdaf_verify_key_init, task_id, vdaf_type),
+            collector_hpke_config: collector_hpke_config.clone(),
+        })
+    }
+}
+
+impl ReportMetadata {
+    /// Does this metatdata have a taskprov extension and does it match the specified id?
+    pub fn is_taskprov(&self, task_id: &Id) -> bool {
+        return self.extensions.iter().any(|x| match x {
+            Extension::Taskprov { payload } => *task_id == compute_task_id(payload),
+            _ => false,
+        });
+    }
+}
diff --git a/daphne/src/vdaf/mod.rs b/daphne/src/vdaf/mod.rs
index c8a6ff3..162c4e2 100644
--- a/daphne/src/vdaf/mod.rs
+++ b/daphne/src/vdaf/mod.rs
@@ -8,8 +8,8 @@ use crate::{
     hpke::HpkeDecrypter,
     messages::{
         encode_u32_bytes, AggregateContinueReq, AggregateInitializeReq, AggregateResp,
-        BatchSelector, HpkeCiphertext, HpkeConfig, Id, PartialBatchSelector, Report, ReportId,
-        ReportMetadata, ReportShare, Time, Transition, TransitionFailure, TransitionVar,
+        BatchSelector, Extension, HpkeCiphertext, HpkeConfig, Id, PartialBatchSelector, Report,
+        ReportId, ReportMetadata, ReportShare, Time, Transition, TransitionFailure, TransitionVar,
     },
     vdaf::{
         prio2::{
@@ -146,20 +146,23 @@ impl VdafConfig {
     /// * `task_id` is the DAP task for which this report is being generated.
     ///
     /// * `measurement` is the measurement.
+    ///
+    /// * `extensions` are the extensions.
     //
     // TODO(issue #100): Truncate the timestamp, as required in DAP-02.
-    pub fn produce_report(
+    pub fn produce_report_with_extensions(
         &self,
         hpke_config_list: &[HpkeConfig],
         time: Time,
         task_id: &Id,
         measurement: DapMeasurement,
+        extensions: Vec<Extension>,
     ) -> Result<Report, DapError> {
         let mut rng = thread_rng();
         let metadata = ReportMetadata {
             id: ReportId(rng.gen()),
             time,
-            extensions: Vec::new(),
+            extensions,
         };
 
         let public_share = Vec::new();
@@ -211,6 +214,37 @@ impl VdafConfig {
         })
     }
 
+    /// Generate a report for a measurement. This method is run by the Client.
+    ///
+    /// # Inputs
+    ///
+    /// * `hpke_config_list` is the sequence of HPKE configs, the first belonging to the Leader and the
+    /// remainder belonging to the Helpers. Note that the current draft only supports one Helper,
+    /// so this method will return an error if `hpke_config_list.len() != 2`.
+    ///
+    /// * `now` is the number of seconds since the UNIX epoch. It is the caller's responsibility to
+    /// ensure this value is truncated to the nearest `min_batch_duration`, as required by the
+    /// spec.
+    ///
+    /// * `task_id` is the DAP task for which this report is being generated.
+    ///
+    /// * `measurement` is the measurement.
+    pub fn produce_report(
+        &self,
+        hpke_config_list: &[HpkeConfig],
+        time: Time,
+        task_id: &Id,
+        measurement: DapMeasurement,
+    ) -> Result<Report, DapError> {
+        self.produce_report_with_extensions(
+            hpke_config_list,
+            time,
+            task_id,
+            measurement,
+            Vec::new(),
+        )
+    }
+
     /// Consume an encrypted input share sent in a report by the Client and return the Prepare
     /// step. This is run by an Aggregator.
     ///
diff --git a/daphne_worker/src/config.rs b/daphne_worker/src/config.rs
index c5aa889..fb5ccad 100644
--- a/daphne_worker/src/config.rs
+++ b/daphne_worker/src/config.rs
@@ -43,6 +43,21 @@ pub(crate) const KV_KEY_PREFIX_BEARER_TOKEN_COLLECTOR: &str = "bearer_token/coll
 pub(crate) const KV_KEY_PREFIX_TASK_CONFIG: &str = "config/task";
 pub(crate) const KV_BINDING_DAP_CONFIG: &str = "DAP_CONFIG";
 
+/// Long-lived parameters for tasks using draft-wang-ppm-dap-taskprov-00 ("taskprov").
+pub(crate) struct TaskprovConfig {
+    /// HPKE collector configuration for all taskprov tasks.
+    pub(crate) hpke_collector_config: HpkeConfig,
+
+    /// VDAF verify key init secret, used to generate the VDAF verification key for a taskprov task.
+    pub(crate) vdaf_verify_key_init: Vec<u8>,
+
+    /// Leader bearer token for all taskprov tasks
+    pub(crate) leader_bearer_token: BearerToken,
+
+    /// Collector bearer token for all taskprov tasks
+    pub(crate) collector_bearer_token: BearerToken,
+}
+
 /// Long-lived parameters used a daphne across DAP tasks.
 pub(crate) struct DaphneWorkerConfig<D> {
     // TODO Determine if we actually need the route countext, and if not, replace it with `Env`.
@@ -84,6 +99,9 @@ pub(crate) struct DaphneWorkerConfig<D> {
 
     /// Indicates if DaphneWorker is used as the Leader.
     is_leader: bool,
+
+    /// draft-wang-ppm-dap-taskprov-00 configuration (optional).
+    pub(crate) taskprov_config: Option<TaskprovConfig>,
 }
 
 impl<D> DaphneWorkerConfig<D> {
@@ -149,6 +167,35 @@ impl<D> DaphneWorkerConfig<D> {
             None
         };
 
+        let taskprov_config = if global_config.allow_taskprov {
+            let hpke_collector_config = serde_json::from_str(
+                ctx.var("DAP_TASKPROV_HPKE_COLLECTOR_CONFIG")?
+                    .to_string()
+                    .as_ref(),
+            )?;
+
+            let vdaf_verify_key_init =
+                hex::decode(ctx.secret("DAP_TASKPROV_VDAF_VERIFY_KEY_INIT")?.to_string())
+                    .map_err(int_err)?;
+
+            let leader_bearer_token =
+                BearerToken::from(ctx.secret("DAP_TASKPROV_LEADER_BEARER_TOKEN")?.to_string());
+
+            let collector_bearer_token = BearerToken::from(
+                ctx.secret("DAP_TASKPROV_COLLECTOR_BEARER_TOKEN")?
+                    .to_string(),
+            );
+
+            Some(TaskprovConfig {
+                hpke_collector_config,
+                vdaf_verify_key_init,
+                leader_bearer_token,
+                collector_bearer_token,
+            })
+        } else {
+            None
+        };
+
         Ok(Self {
             ctx: Some(ctx),
             client,
@@ -162,6 +209,7 @@ impl<D> DaphneWorkerConfig<D> {
             report_shard_count,
             base_url,
             is_leader,
+            taskprov_config,
         })
     }
 
@@ -295,6 +343,16 @@ impl<D> DaphneWorkerConfig<D> {
         .await
     }
 
+    /// Set a leader bearer token for the given task.
+    pub(crate) async fn set_leader_bearer_token(
+        &self,
+        task_id: &Id,
+        token: &BearerToken,
+    ) -> Result<Option<BearerToken>> {
+        self.kv_set_if_not_exists(KV_KEY_PREFIX_BEARER_TOKEN_LEADER, task_id, token.clone())
+            .await
+    }
+
     /// Retrieve from KV the Collector's bearer token for the given task.
     pub(crate) async fn get_collector_bearer_token<'a>(
         &'a self,
@@ -320,6 +378,16 @@ impl<D> DaphneWorkerConfig<D> {
             .await
     }
 
+    /// Define a task in KV
+    pub(crate) async fn set_task_config(
+        &self,
+        task_id: &Id,
+        task_config: &DapTaskConfig,
+    ) -> Result<Option<DapTaskConfig>> {
+        self.kv_set_if_not_exists(KV_KEY_PREFIX_TASK_CONFIG, task_id, task_config.clone())
+            .await
+    }
+
     /// Try retrieving from KV the configuration for the given task. Return an error if the
     /// indicated task is not recognized.
     pub(crate) async fn try_get_task_config<'srv, 'req>(
diff --git a/daphne_worker/src/dap.rs b/daphne_worker/src/dap.rs
index a610f63..dc41674 100644
--- a/daphne_worker/src/dap.rs
+++ b/daphne_worker/src/dap.rs
@@ -50,8 +50,9 @@ use daphne::{
         ReportId, ReportMetadata, TransitionFailure,
     },
     roles::{DapAggregator, DapAuthorizedSender, DapHelper, DapLeader},
+    taskprov::{bad_request, get_taskprov_task_config},
     DapAggregateShare, DapBatchBucket, DapCollectJob, DapError, DapGlobalConfig, DapHelperState,
-    DapOutputShare, DapQueryConfig, DapRequest, DapResponse,
+    DapOutputShare, DapQueryConfig, DapRequest, DapResponse, DapTaskConfig, DapVersion,
 };
 use futures::future::try_join_all;
 use prio::codec::{Decode, Encode};
@@ -215,6 +216,22 @@ impl<'srv, D> BearerTokenProvider<'srv> for DaphneWorkerConfig<D> {
             .await
             .map_err(dap_err)
     }
+
+    fn is_taskprov_leader_bearer_token(&self, token: &BearerToken) -> bool {
+        self.get_global_config().allow_taskprov
+            && match &self.taskprov_config {
+                Some(config) => config.leader_bearer_token == *token,
+                None => false,
+            }
+    }
+
+    fn is_taskprov_collector_bearer_token(&self, token: &BearerToken) -> bool {
+        self.get_global_config().allow_taskprov
+            && match &self.taskprov_config {
+                Some(config) => config.collector_bearer_token == *token,
+                None => false,
+            }
+    }
 }
 
 #[async_trait(?Send)]
@@ -251,11 +268,78 @@ where
         &self.global_config
     }
 
-    async fn get_task_config_for(
+    /// Get an existing task (whether an ordinary task or a previously created
+    /// taskprov task).  If we can't find it, see if there is a taskprov extension
+    /// in the report, and if so create the task.
+    async fn get_task_config_considering_taskprov(
         &'srv self,
         task_id: Cow<'req, Id>,
+        metadata: Option<&ReportMetadata>,
     ) -> std::result::Result<Option<GuardedDapTaskConfig<'req>>, DapError> {
-        self.get_task_config(task_id).await.map_err(dap_err)
+        let found = self
+            .get_task_config(task_id.clone())
+            .await
+            .map_err(dap_err)?;
+        if found.is_some() {
+            return Ok(found);
+        }
+        // Not found and no error.
+        if metadata.is_none() {
+            // No report metadata, so we're not going to find anything.
+            return Ok(None);
+        }
+        let metadata_ref = metadata.unwrap();
+        let taskprov_task_config = get_taskprov_task_config(task_id.as_ref(), metadata_ref)?;
+        if taskprov_task_config.is_some() {
+            let global_config = self.get_global_config();
+            if !global_config.allow_taskprov {
+                // TODO(bhalleycf) if DAP gets a generic denied error, we should use it here.
+                return Err(bad_request("taskprov is not allowed"));
+            }
+            let taskprov_config = self
+                .taskprov_config
+                .as_ref()
+                .ok_or_else(|| DapError::fatal("taskprov configuration not found"))?;
+
+            // We currently always opt-in; any additional checks that would lead to
+            // us opting out should be done here.
+
+            let taskprov_task_id = task_id.as_ref().clone();
+            let task_config = DapTaskConfig::try_from_taskprov(
+                DapVersion::Draft02,
+                &taskprov_task_id,
+                taskprov_task_config.unwrap(),
+                taskprov_config.vdaf_verify_key_init.as_ref(),
+                taskprov_config.hpke_collector_config.as_ref(),
+            )?;
+
+            // Write the leader bearer token to the KV.  We do this so authorize_with_bearer_token()
+            // finds something.
+            //
+            // TODO(bhalleycf) Note that this is generating KV garbage that will
+            // need collection at some point.
+            self.set_leader_bearer_token(
+                &taskprov_task_id,
+                taskprov_config.leader_bearer_token.as_ref(),
+            )
+            .await
+            .map_err(dap_err)?;
+
+            // Write the task config to the KV.
+            //
+            // TODO(bhalleycf) Note that this is generating KV garbage that will
+            // need collection at some point.
+            self.set_task_config(&taskprov_task_id, &task_config)
+                .await
+                .map_err(dap_err)?;
+
+            // Do the usual get again so we cache and return the right type.
+            self.get_task_config(Cow::Owned(taskprov_task_id))
+                .await
+                .map_err(dap_err)
+        } else {
+            Ok(None)
+        }
     }
 
     fn get_current_time(&self) -> u64 {


Test patch:
diff --git a/daphne/src/messages/mod_test.rs b/daphne/src/messages/mod_test.rs
index 5b6901b..a76bbea 100644
--- a/daphne/src/messages/mod_test.rs
+++ b/daphne/src/messages/mod_test.rs
@@ -10,6 +10,7 @@ use crate::messages::{
     HpkeCiphertext, HpkeConfig, HpkeKdfId, HpkeKemId, Id, PartialBatchSelector, Report, ReportId,
     ReportMetadata, ReportShare, Transition, TransitionVar,
 };
+use crate::taskprov::compute_task_id;
 use prio::codec::{Decode, Encode};
 
 #[test]
@@ -232,7 +233,7 @@ fn read_task_config() {
     );
 
     assert_eq!(
-        task_config.compute_task_id().to_hex(),
+        compute_task_id(&task_config.get_encoded()).to_hex(),
         "b4769bb063a8b3312af74297f30fdbf8e0b71c2eb2481f591d1d7de66a4ce34f"
     );
 }
diff --git a/daphne/src/roles_test.rs b/daphne/src/roles_test.rs
index aeb90fb..5a6d09e 100644
--- a/daphne/src/roles_test.rs
+++ b/daphne/src/roles_test.rs
@@ -70,6 +70,7 @@ impl Test {
             min_batch_interval_start: 259200,
             max_batch_interval_end: 259200,
             supported_hpke_kems: vec![HpkeKemId::X25519HkdfSha256],
+            allow_taskprov: true,
         };
 
         // Task Parameters that the Leader and Helper must agree on.
diff --git a/daphne/src/taskprov_test.rs b/daphne/src/taskprov_test.rs
new file mode 100644
index 0000000..77e4547
--- /dev/null
+++ b/daphne/src/taskprov_test.rs
@@ -0,0 +1,30 @@
+// Copyright (c) 2022 Cloudflare, Inc. All rights reserved.
+// SPDX-License-Identifier: BSD-3-Clause
+
+use crate::{
+    messages::taskprov::VdafType, messages::Id, taskprov::compute_vdaf_verify_key,
+    vdaf::VdafVerifyKey,
+};
+
+#[test]
+fn check_vdaf_key_computation() {
+    let task_id = Id([
+        0xb4, 0x76, 0x9b, 0xb0, 0x63, 0xa8, 0xb3, 0x31, 0x2a, 0xf7, 0x42, 0x97, 0xf3, 0x0f, 0xdb,
+        0xf8, 0xe0, 0xb7, 0x1c, 0x2e, 0xb2, 0x48, 0x1f, 0x59, 0x1d, 0x1d, 0x7d, 0xe6, 0x6a, 0x4c,
+        0xe3, 0x4f,
+    ]);
+    let verify_key_init: [u8; 32] = [
+        0x1a, 0x2a, 0x3f, 0x1b, 0xeb, 0xb4, 0xbb, 0xe4, 0x55, 0xea, 0xac, 0xee, 0x29, 0x1a, 0x0f,
+        0x32, 0xd7, 0xe1, 0xbc, 0x6c, 0x75, 0x10, 0x05, 0x60, 0x7b, 0x81, 0xda, 0xc3, 0xa7, 0xda,
+        0x76, 0x1d,
+    ];
+    let vk = compute_vdaf_verify_key(&verify_key_init, &task_id, VdafType::Prio3Aes128Count);
+    let expected: [u8; 16] = [
+        0x57, 0x7a, 0x80, 0x3c, 0xb7, 0xac, 0xa7, 0x5f, 0xdf, 0xb5, 0x86, 0xa9, 0xc1, 0xb6, 0x01,
+        0x75,
+    ];
+    match &vk {
+        VdafVerifyKey::Prio3(bytes) => assert_eq!(*bytes, expected),
+        _ => assert!(false),
+    }
+}
diff --git a/daphne/src/testing.rs b/daphne/src/testing.rs
index bb9a0f4..582645a 100644
--- a/daphne/src/testing.rs
+++ b/daphne/src/testing.rs
@@ -204,6 +204,14 @@ impl<'a> BearerTokenProvider<'a> for MockAggregator {
             ))
         }
     }
+
+    fn is_taskprov_leader_bearer_token(&self, _token: &BearerToken) -> bool {
+        panic!("MockAggregator does not implement the taskprov extension");
+    }
+
+    fn is_taskprov_collector_bearer_token(&self, _token: &BearerToken) -> bool {
+        panic!("MockAggregator does not implement the taskprov extension");
+    }
 }
 
 #[async_trait(?Send)]
@@ -281,9 +289,10 @@ where
         &self.global_config
     }
 
-    async fn get_task_config_for(
+    async fn get_task_config_considering_taskprov(
         &'srv self,
         task_id: Cow<'req, Id>,
+        _metadata: Option<&ReportMetadata>,
     ) -> Result<Option<&'req DapTaskConfig>, DapError> {
         Ok(self.tasks.get(task_id.as_ref()))
     }
diff --git a/daphne_worker_test/tests/e2e.rs b/daphne_worker_test/tests/e2e.rs
index 4a66d53..a49810d 100644
--- a/daphne_worker_test/tests/e2e.rs
+++ b/daphne_worker_test/tests/e2e.rs
@@ -8,17 +8,22 @@ mod test_runner;
 use daphne::{
     constants,
     messages::{
-        BatchSelector, CollectReq, CollectResp, HpkeCiphertext, Id, Interval, Query, Report,
-        ReportId, ReportMetadata,
+        taskprov::{
+            DpConfig, DpMechanism, QueryConfig, QueryConfigVar, TaskConfig, UrlBytes, VdafConfig,
+            VdafTypeVar,
+        },
+        BatchSelector, CollectReq, CollectResp, Extension, HpkeCiphertext, Id, Interval, Query,
+        Report, ReportId, ReportMetadata,
     },
-    DapAggregateResult, DapMeasurement,
+    taskprov::compute_task_id,
+    DapAggregateResult, DapMeasurement, DapTaskConfig, DapVersion,
 };
 use daphne_worker::DaphneWorkerReportSelector;
 use prio::codec::{Decode, Encode};
 use rand::prelude::*;
 use serde::Deserialize;
 use serde_json::json;
-use test_runner::TestRunner;
+use test_runner::{TestRunner, MIN_BATCH_SIZE, TIME_PRECISION};
 
 #[derive(Deserialize)]
 struct InternalTestEndpointForTaskResult {
@@ -276,6 +281,157 @@ async fn e2e_leader_upload() {
         "unexpected response status: {:?}",
         resp.text().await.unwrap()
     );
+
+    // Generate and upload a report with taskprov.
+    //
+    // We have to make this by hand as if we cut and paste a pre-serialized one it
+    // will have an expiring task.
+    let taskprov_task_config = TaskConfig {
+        task_info: "Hi".as_bytes().to_vec(),
+        aggregator_endpoints: vec![
+            UrlBytes {
+                bytes: "https://test1".as_bytes().to_vec(),
+            },
+            UrlBytes {
+                bytes: "https://test2".as_bytes().to_vec(),
+            },
+        ],
+        query_config: QueryConfig {
+            time_precision: 0x01,
+            max_batch_query_count: 128,
+            min_batch_size: 1024,
+            var: QueryConfigVar::FixedSize {
+                max_batch_size: 2048,
+            },
+        },
+        task_expiration: t.now + 86400,
+        vdaf_config: VdafConfig {
+            dp_config: DpConfig {
+                mechanism: DpMechanism::None,
+            },
+            var: VdafTypeVar::Prio3Aes128Count,
+        },
+    };
+    let payload = taskprov_task_config.get_encoded();
+    let task_id = compute_task_id(&payload);
+    let extensions = vec![Extension::Taskprov { payload }];
+    let report = t
+        .task_config
+        .vdaf
+        .produce_report_with_extensions(
+            &hpke_config_list,
+            t.now,
+            &task_id,
+            DapMeasurement::U64(23),
+            extensions,
+        )
+        .unwrap();
+    t.leader_post_expect_ok(
+        &client,
+        path,
+        constants::MEDIA_TYPE_REPORT,
+        report.get_encoded(),
+    )
+    .await;
+
+    // Generate and upload a report with taskprov but with the wrong id
+    let payload = taskprov_task_config.get_encoded();
+    let mut bad_payload = payload.clone();
+    bad_payload[0] = u8::wrapping_add(bad_payload[0], 1);
+    let task_id = compute_task_id(&bad_payload);
+    let extensions = vec![Extension::Taskprov { payload }];
+    let report = t
+        .task_config
+        .vdaf
+        .produce_report_with_extensions(
+            &hpke_config_list,
+            t.now,
+            &task_id,
+            DapMeasurement::U64(23),
+            extensions,
+        )
+        .unwrap();
+    t.leader_post_expect_abort(
+        &client,
+        None, // dap_auth_token
+        path,
+        constants::MEDIA_TYPE_REPORT,
+        report.get_encoded(),
+        400,
+        "unrecognizedTask",
+    )
+    .await;
+
+    // Generate and upload a report with two copies of the taskprov extension
+    let payload = taskprov_task_config.get_encoded();
+    let task_id = compute_task_id(&payload);
+    let extensions = vec![
+        Extension::Taskprov {
+            payload: payload.clone(),
+        },
+        Extension::Taskprov { payload },
+    ];
+    let report = t
+        .task_config
+        .vdaf
+        .produce_report_with_extensions(
+            &hpke_config_list,
+            t.now,
+            &task_id,
+            DapMeasurement::U64(23),
+            extensions,
+        )
+        .unwrap();
+    t.leader_post_expect_abort(
+        &client,
+        None, // dap_auth_token
+        path,
+        constants::MEDIA_TYPE_REPORT,
+        report.get_encoded(),
+        400,
+        "unrecognizedMessage",
+    )
+    .await;
+
+    // Generate and upload a report with taskprov but with only one aggregator endpoint (an error).
+    // (This report is also expired, but we notice the endpoint problem first).
+    let payload = [
+        0x02u8, 0x48, 0x69, 0x00, 0x0e, 0x00, 0x0c, 0x68, 0x74, 0x74, 0x70, 0x73, 0x3a, 0x2f, 0x2f,
+        0x74, 0x65, 0x73, 0x74, 0x02, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x80,
+        0x00, 0x00, 0x04, 0x00, 0x00, 0x00, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x63, 0x52, 0xf9,
+        0xa5, 0x01, 0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x18, 0x01, 0x02, 0x03, 0x04, 0x04, 0x03,
+        0x02, 0x01, 0x02, 0x02, 0x03, 0x04, 0x04, 0x03, 0x02, 0x02, 0x03, 0x02, 0x03, 0x04, 0x04,
+        0x03, 0x02, 0x03,
+    ];
+    let extensions = vec![Extension::Taskprov {
+        payload: payload.to_vec(),
+    }];
+    let task_id = Id([
+        0xb4, 0x76, 0x9b, 0xb0, 0x63, 0xa8, 0xb3, 0x31, 0x2a, 0xf7, 0x42, 0x97, 0xf3, 0x0f, 0xdb,
+        0xf8, 0xe0, 0xb7, 0x1c, 0x2e, 0xb2, 0x48, 0x1f, 0x59, 0x1d, 0x1d, 0x7d, 0xe6, 0x6a, 0x4c,
+        0xe3, 0x4f,
+    ]);
+    let report = t
+        .task_config
+        .vdaf
+        .produce_report_with_extensions(
+            &hpke_config_list,
+            t.now,
+            &task_id,
+            DapMeasurement::U64(23),
+            extensions,
+        )
+        .unwrap();
+    t.leader_post_expect_abort(
+        &client,
+        None, // dap_auth_token
+        path,
+        constants::MEDIA_TYPE_REPORT,
+        report.get_encoded(),
+        400,
+        "badRequest",
+    )
+    .await;
 }
 
 #[tokio::test]
@@ -941,3 +1097,147 @@ async fn e2e_fixed_size() {
     )
     .await;
 }
+
+#[tokio::test]
+#[cfg_attr(not(feature = "test_e2e"), ignore)]
+async fn e2e_leader_collect_taskprov_ok() {
+    let t = TestRunner::default().await;
+    let batch_interval = t.batch_interval();
+
+    let client = t.http_client();
+    let hpke_config_list = t.get_hpke_configs(&client).await;
+
+    let taskprov_task_config = TaskConfig {
+        task_info: "".as_bytes().to_vec(),
+        aggregator_endpoints: vec![
+            UrlBytes {
+                bytes: t.task_config.leader_url.as_str().as_bytes().to_vec(),
+            },
+            UrlBytes {
+                bytes: t.task_config.helper_url.as_str().as_bytes().to_vec(),
+            },
+        ],
+        query_config: QueryConfig {
+            time_precision: TIME_PRECISION,
+            max_batch_query_count: 65535,
+            min_batch_size: u32::try_from(MIN_BATCH_SIZE).unwrap(),
+            var: QueryConfigVar::TimeInterval,
+        },
+        task_expiration: t.now + 86400 * 14,
+        vdaf_config: VdafConfig {
+            dp_config: DpConfig {
+                mechanism: DpMechanism::None,
+            },
+            var: VdafTypeVar::Prio3Aes128Sum { bit_length: 10 },
+        },
+    };
+    let payload = taskprov_task_config.get_encoded();
+    let task_id = compute_task_id(&payload);
+    let task_config = DapTaskConfig::try_from_taskprov(
+        DapVersion::Draft02,
+        &task_id.clone(),
+        taskprov_task_config.clone(),
+        &t.taskprov_vdaf_verify_key_init,
+        &t.taskprov_collector_hpke_receiver.config,
+    )
+    .unwrap();
+
+    // The reports are uploaded in the background.
+    let mut rng = thread_rng();
+    for _ in 0..t.task_config.min_batch_size {
+        let extensions = vec![Extension::Taskprov {
+            payload: payload.clone(),
+        }];
+        let now = rng.gen_range(batch_interval.start..batch_interval.end());
+        t.leader_post_expect_ok(
+            &client,
+            "upload",
+            constants::MEDIA_TYPE_REPORT,
+            task_config
+                .vdaf
+                .produce_report_with_extensions(
+                    &hpke_config_list,
+                    now,
+                    &task_id,
+                    DapMeasurement::U64(1),
+                    extensions,
+                )
+                .unwrap()
+                .get_encoded(),
+        )
+        .await;
+    }
+
+    // Get the collect URI.
+    let collect_req = CollectReq {
+        task_id: task_id.clone(),
+        query: Query::TimeInterval {
+            batch_interval: batch_interval.clone(),
+        },
+        agg_param: Vec::new(),
+    };
+    let collect_uri = t
+        .leader_post_collect_using_token(
+            &client,
+            collect_req.get_encoded(),
+            &"I am the collector!".to_string(), // Keep in sync with wrangler.toml
+        )
+        .await;
+    println!("collect_uri: {}", collect_uri);
+
+    // Poll the collect URI before the CollectResp is ready.
+    let resp = client.get(collect_uri.as_str()).send().await.unwrap();
+    assert_eq!(resp.status(), 202, "response: {:?}", resp);
+
+    // The reports are aggregated in the background.
+    let agg_telem = t
+        .internal_process(
+            &client,
+            &DaphneWorkerReportSelector {
+                max_agg_jobs: 100, // Needs to be sufficiently large to touch each bucket.
+                max_reports: 100,
+            },
+        )
+        .await;
+    assert_eq!(
+        agg_telem.reports_processed, task_config.min_batch_size,
+        "reports processed"
+    );
+    assert_eq!(
+        agg_telem.reports_aggregated, task_config.min_batch_size,
+        "reports aggregated"
+    );
+    assert_eq!(
+        agg_telem.reports_collected, task_config.min_batch_size,
+        "reports collected"
+    );
+
+    // Poll the collect URI.
+    let resp = client.get(collect_uri.as_str()).send().await.unwrap();
+    assert_eq!(resp.status(), 200);
+
+    let collect_resp = CollectResp::get_decoded(&resp.bytes().await.unwrap()).unwrap();
+    let agg_res = task_config
+        .vdaf
+        .consume_encrypted_agg_shares(
+            &t.taskprov_collector_hpke_receiver,
+            &task_id,
+            &BatchSelector::TimeInterval {
+                batch_interval: batch_interval.clone(),
+            },
+            collect_resp.report_count,
+            collect_resp.encrypted_agg_shares.clone(),
+        )
+        .await
+        .unwrap();
+    assert_eq!(
+        agg_res,
+        DapAggregateResult::U128(task_config.min_batch_size as u128)
+    );
+
+    // Poll the collect URI once more. Expect the response to be the same as the first, per HTTP
+    // GET semantics.
+    let resp = client.get(collect_uri.as_str()).send().await.unwrap();
+    assert_eq!(resp.status(), 200);
+    assert_eq!(resp.bytes().await.unwrap(), collect_resp.get_encoded());
+}
diff --git a/daphne_worker_test/tests/test_runner.rs b/daphne_worker_test/tests/test_runner.rs
index 24b070f..0e077ed 100644
--- a/daphne_worker_test/tests/test_runner.rs
+++ b/daphne_worker_test/tests/test_runner.rs
@@ -7,7 +7,7 @@ use assert_matches::assert_matches;
 use daphne::{
     constants::MEDIA_TYPE_COLLECT_REQ,
     hpke::HpkeReceiverConfig,
-    messages::{Duration, HpkeConfig, HpkeKemId, Id, Interval},
+    messages::{Duration, HpkeAeadId, HpkeConfig, HpkeKdfId, HpkeKemId, Id, Interval},
     DapGlobalConfig, DapLeaderProcessTelemetry, DapQueryConfig, DapTaskConfig, DapVersion,
     Prio3Config, VdafConfig,
 };
@@ -24,9 +24,9 @@ use tokio::task::JoinHandle;
 use url::Url;
 
 const VDAF_CONFIG: &VdafConfig = &VdafConfig::Prio3(Prio3Config::Sum { bits: 10 });
-const MIN_BATCH_SIZE: u64 = 10;
-const MAX_BATCH_SIZE: u64 = 12;
-const TIME_PRECISION: Duration = 3600; // seconds
+pub(crate) const MIN_BATCH_SIZE: u64 = 10;
+pub(crate) const MAX_BATCH_SIZE: u64 = 12;
+pub(crate) const TIME_PRECISION: Duration = 3600; // seconds
 
 #[derive(Deserialize)]
 struct InternalTestAddTaskResult {
@@ -46,6 +46,8 @@ pub struct TestRunner {
     pub leader_bearer_token: String,
     pub collector_bearer_token: String,
     pub collector_hpke_receiver: HpkeReceiverConfig,
+    pub taskprov_vdaf_verify_key_init: Vec<u8>,
+    pub taskprov_collector_hpke_receiver: HpkeReceiverConfig,
 }
 
 #[allow(dead_code)]
@@ -101,14 +103,27 @@ impl TestRunner {
             collector_hpke_config: collector_hpke_receiver.config.clone(),
         };
 
-        // This needs to be kept in-sync with daphne_worker_test/wrangler.toml.
+        // This block needs to be kept in-sync with daphne_worker_test/wrangler.toml.
         let global_config = DapGlobalConfig {
             report_storage_epoch_duration: 604800,
             max_batch_duration: 360000,
             min_batch_interval_start: 259200,
             max_batch_interval_end: 259200,
             supported_hpke_kems: vec![HpkeKemId::X25519HkdfSha256],
+            allow_taskprov: true,
         };
+        let taskprov_vdaf_verify_key_init =
+            hex::decode("0074a5dd6e9dac501f73f7a961193b2b").unwrap();
+        let taskprov_collector_hpke_receiver = HpkeReceiverConfig::new(
+            HpkeConfig {
+                id: 23,
+                kem_id: HpkeKemId::P256HkdfSha256,
+                kdf_id: HpkeKdfId::HkdfSha256,
+                aead_id: HpkeAeadId::Aes128Gcm,
+                public_key: hex::decode("047dab625e0d269abcc28c611bebf5a60987ddf7e23df0e0aa343e5774ad81a1d0160d9252b82b4b5c52354205f5ec945645cb79facff8d85c9c31b490cdf35466").unwrap()
+            },
+            hex::decode("9ce9851512df3ea674b108b305c3f8c424955a94d93fd53ecf3c3f17f7d1df9e").unwrap()
+        );
 
         let leader_bearer_token = hex::encode(&rng.gen::<[u8; 16]>());
         let collector_bearer_token = hex::encode(&rng.gen::<[u8; 16]>());
@@ -122,6 +137,8 @@ impl TestRunner {
             leader_bearer_token,
             collector_bearer_token,
             collector_hpke_receiver,
+            taskprov_vdaf_verify_key_init,
+            taskprov_collector_hpke_receiver,
         };
 
         let vdaf_verify_key_base64url = base64::encode_config(
@@ -312,7 +329,12 @@ impl TestRunner {
         );
     }
 
-    pub async fn leader_post_collect(&self, client: &reqwest::Client, data: Vec<u8>) -> Url {
+    pub async fn leader_post_collect_using_token(
+        &self,
+        client: &reqwest::Client,
+        data: Vec<u8>,
+        token: &String,
+    ) -> Url {
         let url = self.leader_url.join("collect").unwrap();
         let mut headers = reqwest::header::HeaderMap::new();
         headers.insert(
@@ -321,7 +343,7 @@ impl TestRunner {
         );
         headers.insert(
             reqwest::header::HeaderName::from_static("dap-auth-token"),
-            reqwest::header::HeaderValue::from_str(&self.collector_bearer_token).unwrap(),
+            reqwest::header::HeaderValue::from_str(token).unwrap(),
         );
         let resp = client
             .post(url.as_str())
@@ -341,6 +363,11 @@ impl TestRunner {
         collect_uri.parse().unwrap()
     }
 
+    pub async fn leader_post_collect(&self, client: &reqwest::Client, data: Vec<u8>) -> Url {
+        self.leader_post_collect_using_token(client, data, &self.collector_bearer_token)
+            .await
+    }
+
     #[allow(dead_code)]
     pub async fn internal_process(
         &self,
diff --git a/daphne_worker_test/wrangler.toml b/daphne_worker_test/wrangler.toml
index c167c12..097fd5c 100644
--- a/daphne_worker_test/wrangler.toml
+++ b/daphne_worker_test/wrangler.toml
@@ -31,9 +31,21 @@ DAP_GLOBAL_CONFIG = """{
      "max_batch_duration": 360000,
      "min_batch_interval_start": 259200,
      "max_batch_interval_end": 259200,
-     "supported_hpke_kems": ["x25519_hkdf_sha256"]
+     "supported_hpke_kems": ["x25519_hkdf_sha256"],
+     "allow_taskprov": true
 }"""
-
+DAP_TASKPROV_HPKE_COLLECTOR_CONFIG = """{
+  "id": 23,
+  "kem_id": "p256_hkdf_sha256",
+  "kdf_id": "hkdf_sha256",
+  "aead_id": "aes128_gcm",
+  "public_key": "047dab625e0d269abcc28c611bebf5a60987ddf7e23df0e0aa343e5774ad81a1d0160d9252b82b4b5c52354205f5ec945645cb79facff8d85c9c31b490cdf35466"
+}"""
+# The secret key for the HPKE collector public key, recorded here for testing convenience:
+#     9ce9851512df3ea674b108b305c3f8c424955a94d93fd53ecf3c3f17f7d1df9e   # SECRET
+DAP_TASKPROV_VDAF_VERIFY_KEY_INIT = "0074a5dd6e9dac501f73f7a961193b2b" # SECRET
+DAP_TASKPROV_LEADER_BEARER_TOKEN = "I am the leader!" # SECRET
+DAP_TASKPROV_COLLECTOR_BEARER_TOKEN = "I am the collector!" # SECRET
 
 [env.leader.durable_objects]
 bindings = [
@@ -74,9 +86,21 @@ DAP_GLOBAL_CONFIG = """{
   "max_batch_duration": 360000,
   "min_batch_interval_start": 259200,
   "max_batch_interval_end": 259200,
-  "supported_hpke_kems": ["x25519_hkdf_sha256"]
+  "supported_hpke_kems": ["x25519_hkdf_sha256"],
+  "allow_taskprov": true
 }"""
-
+DAP_TASKPROV_HPKE_COLLECTOR_CONFIG = """{
+  "id": 23,
+  "kem_id": "p256_hkdf_sha256",
+  "kdf_id": "hkdf_sha256",
+  "aead_id": "aes128_gcm",
+  "public_key": "047dab625e0d269abcc28c611bebf5a60987ddf7e23df0e0aa343e5774ad81a1d0160d9252b82b4b5c52354205f5ec945645cb79facff8d85c9c31b490cdf35466"
+}"""
+# The secret key for the HPKE collector public key, recorded here for testing convenience:
+#     9ce9851512df3ea674b108b305c3f8c424955a94d93fd53ecf3c3f17f7d1df9e   # SECRET
+DAP_TASKPROV_VDAF_VERIFY_KEY_INIT = "0074a5dd6e9dac501f73f7a961193b2b" # SECRET
+DAP_TASKPROV_LEADER_BEARER_TOKEN = "I am the leader!" # SECRET
+DAP_TASKPROV_COLLECTOR_BEARER_TOKEN = "I am the collector!" # SECRET
 
 [env.helper.durable_objects]
 bindings = [