Repository: cloudflare/daphne
Pull request #711 (2024-11-04 12:23:14)

Eagerly parse taskprov advertisement
- Rename fields and types that relate to taskprov advertisements to reflect such
- Eagerly deserialize taskprov advertisement



Patch:
diff --git a/crates/dapf/src/acceptance/mod.rs b/crates/dapf/src/acceptance/mod.rs
index 2f61124..5ceb0fd 100644
--- a/crates/dapf/src/acceptance/mod.rs
+++ b/crates/dapf/src/acceptance/mod.rs
@@ -24,8 +24,8 @@ use async_trait::async_trait;
 use daphne::{
     hpke::{HpkeConfig, HpkeKemId, HpkeReceiverConfig},
     messages::{
-        self, AggregateShareReq, AggregationJobId, Base64Encode, BatchId, BatchSelector,
-        PartialBatchSelector, TaskId,
+        self, taskprov::TaskprovAdvertisement, AggregateShareReq, AggregationJobId, Base64Encode,
+        BatchId, BatchSelector, PartialBatchSelector, TaskId,
     },
     metrics::DaphneMetrics,
     roles::DapReportInitializer,
@@ -203,7 +203,7 @@ struct TestTaskConfig {
     pub hpke_config_list: [HpkeConfig; 2],
     pub fake_leader_hpke_receiver_config: HpkeReceiverConfig,
     pub task_config: DapTaskConfig,
-    pub taskprov_advertisement: Option<String>,
+    pub taskprov_advertisement: Option<TaskprovAdvertisement>,
 }
 
 impl Test {
@@ -522,7 +522,7 @@ impl Test {
                 agg_job_init_req,
                 task_config.version,
                 functions::helper::Options {
-                    taskprov_advertisement: taskprov_advertisement.as_deref(),
+                    taskprov_advertisement: taskprov_advertisement.as_ref(),
                     bearer_token: self.bearer_token.as_ref(),
                 },
             )
@@ -554,7 +554,7 @@ impl Test {
         leader_agg_share: DapAggregateShare,
         batch_id: BatchId,
         version: DapVersion,
-        taskprov_advertisement: Option<&str>,
+        taskprov_advertisement: Option<&TaskprovAdvertisement>,
         task_id: TaskId,
     ) -> Result<Duration> {
         // Prepare AggregateShareReq.
@@ -672,7 +672,7 @@ impl Test {
                 leader_agg_share,
                 batch_id,
                 version,
-                taskprov_advertisement.as_deref(),
+                taskprov_advertisement.as_ref(),
                 *task_id,
             )
             .await?;
diff --git a/crates/dapf/src/functions/helper.rs b/crates/dapf/src/functions/helper.rs
index 0dc1669..1392458 100644
--- a/crates/dapf/src/functions/helper.rs
+++ b/crates/dapf/src/functions/helper.rs
@@ -5,7 +5,10 @@ use anyhow::{anyhow, Context as _};
 use daphne::{
     constants::DapMediaType,
     error::aborts::ProblemDetails,
-    messages::{AggregateShareReq, AggregationJobInitReq, AggregationJobResp},
+    messages::{
+        taskprov::TaskprovAdvertisement, AggregateShareReq, AggregationJobInitReq,
+        AggregationJobResp,
+    },
     DapVersion,
 };
 use daphne_service_utils::{bearer_token::BearerToken, http_headers};
@@ -34,6 +37,7 @@ impl HttpClient {
                     .with_context(|| {
                         format!("AggregationJobInitReq media type is not defined for {version}")
                     })?,
+                version,
                 opts,
             )?)
             .send()
@@ -80,6 +84,7 @@ impl HttpClient {
                     .with_context(|| {
                         format!("AggregateShareReq media type is not defined for {version}")
                     })?,
+                version,
                 opts,
             )?)
             .send()
@@ -106,12 +111,13 @@ impl HttpClient {
 
 #[derive(Default, Debug)]
 pub struct Options<'s> {
-    pub taskprov_advertisement: Option<&'s str>,
+    pub taskprov_advertisement: Option<&'s TaskprovAdvertisement>,
     pub bearer_token: Option<&'s BearerToken>,
 }
 
 fn construct_request_headers(
     media_type: &str,
+    version: DapVersion,
     options: Options<'_>,
 ) -> Result<header::HeaderMap, header::InvalidHeaderValue> {
     let mut headers = header::HeaderMap::new();
@@ -126,7 +132,7 @@ fn construct_request_headers(
     if let Some(taskprov) = taskprov_advertisement {
         headers.insert(
             const { header::HeaderName::from_static(http_headers::DAP_TASKPROV) },
-            header::HeaderValue::from_str(taskprov)?,
+            header::HeaderValue::from_str(&taskprov.serialize_to_header_value(version).unwrap())?,
         );
     }
     if let Some(token) = bearer_token {
diff --git a/crates/daphne-server/src/lib.rs b/crates/daphne-server/src/lib.rs
index d0bb6dc..48b7848 100644
--- a/crates/daphne-server/src/lib.rs
+++ b/crates/daphne-server/src/lib.rs
@@ -166,7 +166,7 @@ impl router::DaphneService for App {
     }
 
     async fn is_using_taskprov(&self, req: &daphne::DapRequestMeta) -> Result<bool, DapError> {
-        if req.taskprov.is_some() {
+        if req.taskprov_advertisement.is_some() {
             Ok(true)
         } else if self
             .get_task_config_for(&req.task_id)
diff --git a/crates/daphne-server/src/roles/leader.rs b/crates/daphne-server/src/roles/leader.rs
index 77b50eb..3e0bcb1 100644
--- a/crates/daphne-server/src/roles/leader.rs
+++ b/crates/daphne-server/src/roles/leader.rs
@@ -153,7 +153,7 @@ impl crate::App {
                 .map_err(|e| fatal_error!(err = ?e, "failed to construct content-type header"))?,
         );
 
-        let bearer_token = if meta.taskprov.is_some() {
+        let bearer_token = if meta.taskprov_advertisement.is_some() {
             if let Some(bearer_token) = self
                 .service_config
                 .taskprov
@@ -191,12 +191,13 @@ impl crate::App {
                 .map_err(|e| fatal_error!(err = ?e, "failed to construct authentication header"))?,
         );
 
-        if let Some(taskprov_advertisement) = meta.taskprov.as_deref() {
+        if let Some(taskprov_advertisement) = &meta.taskprov_advertisement {
             headers.insert(
                 HeaderName::from_static(http_headers::DAP_TASKPROV),
-                HeaderValue::from_str(taskprov_advertisement).map_err(
-                    |e| fatal_error!(err = ?e, "failed to construct dap-taskprov header"),
-                )?,
+                HeaderValue::from_str(
+                    &taskprov_advertisement.serialize_to_header_value(meta.version)?,
+                )
+                .map_err(|e| fatal_error!(err = ?e, "failed to construct dap-taskprov header"))?,
             );
         }
 
diff --git a/crates/daphne-server/src/router/extractor.rs b/crates/daphne-server/src/router/extractor.rs
index fd9bf1a..af14bd2 100644
--- a/crates/daphne-server/src/router/extractor.rs
+++ b/crates/daphne-server/src/router/extractor.rs
@@ -13,7 +13,8 @@ use daphne::{
     error::DapAbort,
     fatal_error,
     messages::{
-        request::resource, AggregateShareReq, AggregationJobInitReq, CollectionReq, Report, TaskId,
+        request::resource, taskprov::TaskprovAdvertisement, AggregateShareReq,
+        AggregationJobInitReq, CollectionReq, Report, TaskId,
     },
     DapError, DapRequest, DapRequestMeta, DapVersion,
 };
@@ -185,10 +186,16 @@ where
             ));
         };
 
+        let taskprov_advertisement =
+            extract_header_as_str(&parts.headers, http_headers::DAP_TASKPROV)
+                .map(|h| TaskprovAdvertisement::parse_taskprov_advertisement(h, &task_id, version))
+                .transpose()
+                .map_err(|e| AxumDapResponse::new_error(e, state.server_metrics()))?;
+
         let meta = DapRequestMeta {
             version,
             task_id,
-            taskprov: extract_header_as_string(&parts.headers, http_headers::DAP_TASKPROV),
+            taskprov_advertisement,
             media_type,
         };
 
@@ -249,9 +256,10 @@ where
     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)
+        let bearer_token = extract_header_as_str(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 cf_tls_client_auth =
+            extract_header_as_str(req.headers(), "X-Client-Cert-Verified").map(ToString::to_string);
 
         let request = UnauthenticatedDapRequestExtractor::from_request(req, state)
             .await?
@@ -316,8 +324,8 @@ where
     }
 }
 
-fn extract_header_as_string(headers: &HeaderMap, header: &'static str) -> Option<String> {
-    headers.get(header)?.to_str().ok().map(ToString::to_string)
+fn extract_header_as_str<'s>(headers: &'s HeaderMap, header: &'static str) -> Option<&'s str> {
+    headers.get(header)?.to_str().ok()
 }
 
 #[cfg(test)]
@@ -338,7 +346,10 @@ mod test {
     use daphne::{
         async_test_versions,
         constants::DapMediaType,
-        messages::{request::resource, AggregationJobId, Base64Encode, CollectionJobId, TaskId},
+        messages::{
+            request::resource, taskprov::TaskprovAdvertisement, AggregationJobId, Base64Encode,
+            CollectionJobId, TaskId,
+        },
         DapError, DapRequestMeta, DapSender, DapVersion,
     };
     use daphne_service_utils::{bearer_token::BearerToken, http_headers};
@@ -408,7 +419,7 @@ mod test {
             }
 
             async fn is_using_taskprov(&self, req: &DapRequestMeta) -> Result<bool, DapError> {
-                Ok(req.taskprov.is_some())
+                Ok(req.taskprov_advertisement.is_some())
             }
         }
 
@@ -460,7 +471,14 @@ mod test {
             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),
+            code => {
+                let payload = hyper::body::to_bytes(resp.into_body()).await.unwrap();
+                eprintln!(
+                    "body was: {}",
+                    String::from_utf8_lossy(&payload).into_owned()
+                );
+                Err(code)
+            }
         }
     }
 
@@ -575,11 +593,42 @@ mod test {
     async_test_versions! { missing_auth_is_rejected }
 
     async fn mtls_auth_is_enough(version: DapVersion) {
+        let taskprov_advertisement = TaskprovAdvertisement {
+            task_info: b"cool task".into(),
+            leader_url: daphne::messages::taskprov::UrlBytes {
+                bytes: b"http://leader".into(),
+            },
+            helper_url: daphne::messages::taskprov::UrlBytes {
+                bytes: b"http://helper".into(),
+            },
+            query_config: daphne::messages::taskprov::QueryConfig {
+                time_precision: 1,
+                max_batch_query_count: 1,
+                min_batch_size: 1,
+                var: daphne::messages::taskprov::QueryConfigVar::TimeInterval,
+            },
+            task_expiration: 1,
+            vdaf_config: daphne::messages::taskprov::VdafConfig {
+                dp_config: daphne::messages::taskprov::DpConfig::None,
+                var: daphne::messages::taskprov::VdafTypeVar::Prio2 { dimension: 1 },
+            },
+        };
+
         let req = test::<_, resource::None>(
             Request::builder()
-                .uri(format!("/{version}/{}/auth", mk_task_id().to_base64url()))
+                .uri(format!(
+                    "/{version}/{}/auth",
+                    taskprov_advertisement
+                        .compute_task_id(version)
+                        .to_base64url()
+                ))
                 .header("X-Client-Cert-Verified", "SUCCESS")
-                .header(http_headers::DAP_TASKPROV, "some-taskprov-string")
+                .header(
+                    http_headers::DAP_TASKPROV,
+                    taskprov_advertisement
+                        .serialize_to_header_value(version)
+                        .unwrap(),
+                )
                 .body(Body::empty())
                 .unwrap(),
         )
diff --git a/crates/daphne/src/lib.rs b/crates/daphne/src/lib.rs
index 4e630a6..222f387 100644
--- a/crates/daphne/src/lib.rs
+++ b/crates/daphne/src/lib.rs
@@ -68,13 +68,13 @@ use crate::{
 pub use error::DapError;
 use error::FatalDapError;
 use hpke::{HpkeConfig, HpkeKemId};
-use messages::encode_base64url;
+use messages::taskprov::TaskprovAdvertisement;
 #[cfg(feature = "experimental")]
 use prio::codec::Decode;
 #[cfg(feature = "experimental")]
 use prio::vdaf::poplar1::Poplar1AggregationParam;
 use prio::{
-    codec::{CodecError, Encode, ParameterizedDecode, ParameterizedEncode},
+    codec::{CodecError, Encode, ParameterizedDecode},
     vdaf::Aggregatable as AggregatableTrait,
 };
 pub use protocol::aggregator::ReplayProtection;
@@ -540,14 +540,14 @@ pub struct DapTaskParameters {
 #[cfg(any(test, feature = "test-utils"))]
 impl DapTaskParameters {
     /// Construct a new task config using the taskprov extension. Return the task ID and the
-    /// taskprov advertisement encoded as a base64url string.
+    /// taskprov advertisement
     pub fn to_config_with_taskprov(
         &self,
         task_info: Vec<u8>,
         now: Time,
         taskprov_config: roles::aggregator::TaskprovConfig<'_>,
-    ) -> Result<(DapTaskConfig, TaskId, String), DapError> {
-        let task_config = messages::taskprov::TaskConfig {
+    ) -> Result<(DapTaskConfig, TaskId, TaskprovAdvertisement), DapError> {
+        let taskprov_advertisement = messages::taskprov::TaskprovAdvertisement {
             task_info,
             leader_url: messages::taskprov::UrlBytes {
                 bytes: self.leader_url.to_string().into_bytes(),
@@ -568,16 +568,13 @@ impl DapTaskParameters {
             },
         };
 
-        let encoded_taskprov_config = task_config
-            .get_encoded_with_param(&self.version)
-            .map_err(DapError::encoding)?;
-        let task_id = taskprov::compute_task_id(&encoded_taskprov_config);
+        let task_id = taskprov_advertisement.compute_task_id(self.version);
 
         // Compute the DAP task config.
         let task_config = taskprov::DapTaskConfigNeedsOptIn::try_from_taskprov(
             self.version,
             &task_id,
-            task_config,
+            taskprov_advertisement.clone(),
             taskprov_config,
         )
         .unwrap()
@@ -586,8 +583,6 @@ impl DapTaskParameters {
             num_agg_span_shards: self.num_agg_span_shards,
         });
 
-        let taskprov_advertisement = encode_base64url(&encoded_taskprov_config);
-
         Ok((task_config, task_id, taskprov_advertisement))
     }
 }
@@ -756,12 +751,13 @@ impl DapTaskConfig {
     }
 
     /// Leader: Resolve taskprov advertisement to send in a request to the Helper.
-    pub(crate) fn resolve_taskprove_advertisement(&self) -> Result<Option<String>, DapError> {
+    pub(crate) fn resolve_taskprove_advertisement(
+        &self,
+    ) -> Result<Option<TaskprovAdvertisement>, DapError> {
         if let DapTaskConfigMethod::Taskprov { info: _ } = &self.method {
-            let encoded_taskprov_config = messages::taskprov::TaskConfig::try_from(self)?
-                .get_encoded_with_param(&self.version)
-                .map_err(DapError::encoding)?;
-            Ok(Some(encode_base64url(encoded_taskprov_config)))
+            Ok(Some(messages::taskprov::TaskprovAdvertisement::try_from(
+                self,
+            )?))
         } else {
             Ok(None)
         }
diff --git a/crates/daphne/src/messages/request.rs b/crates/daphne/src/messages/request.rs
index 0b507c1..1808227 100644
--- a/crates/daphne/src/messages/request.rs
+++ b/crates/daphne/src/messages/request.rs
@@ -3,6 +3,7 @@
 
 use std::ops::Deref;
 
+use super::taskprov::TaskprovAdvertisement;
 use crate::{constants::DapMediaType, error::DapAbort, messages::TaskId, DapVersion};
 
 pub mod resource {
@@ -36,7 +37,7 @@ pub struct DapRequestMeta {
     pub task_id: TaskId,
 
     /// taskprov: The task advertisement, sent in the `dap-taskprov` header.
-    pub taskprov: Option<String>,
+    pub taskprov_advertisement: Option<TaskprovAdvertisement>,
 }
 
 impl DapRequestMeta {
diff --git a/crates/daphne/src/messages/taskprov.rs b/crates/daphne/src/messages/taskprov.rs
index 4c7e565..660d956 100644
--- a/crates/daphne/src/messages/taskprov.rs
+++ b/crates/daphne/src/messages/taskprov.rs
@@ -8,7 +8,7 @@ use crate::messages::{
     QUERY_TYPE_TIME_INTERVAL,
 };
 use crate::pine::PineParam;
-use crate::DapVersion;
+use crate::{DapError, DapVersion};
 use prio::codec::{
     decode_u8_items, encode_u8_items, CodecError, Decode, Encode, ParameterizedDecode,
     ParameterizedEncode,
@@ -16,7 +16,11 @@ use prio::codec::{
 use serde::{Deserialize, Serialize};
 use std::io::{Cursor, Read};
 
-use super::{decode_u16_prefixed, encode_u16_prefixed};
+use super::{
+    decode_base64url_vec, decode_u16_prefixed, encode_base64url, encode_u16_prefixed, TaskId,
+};
+use crate::error::DapAbort;
+use crate::taskprov::compute_task_id;
 
 // VDAF type codes.
 const VDAF_TYPE_PRIO2: u32 = 0xFFFF_0000;
@@ -408,7 +412,7 @@ impl ParameterizedDecode<(DapVersion, Option<usize>)> for QueryConfig {
 
 /// A DAP task configuration.
 #[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
-pub struct TaskConfig {
+pub struct TaskprovAdvertisement {
     pub task_info: Vec<u8>,
     pub leader_url: UrlBytes,
     pub helper_url: UrlBytes,
@@ -417,7 +421,47 @@ pub struct TaskConfig {
     pub vdaf_config: VdafConfig,
 }
 
-impl ParameterizedEncode<DapVersion> for TaskConfig {
+impl TaskprovAdvertisement {
+    /// Check for a taskprov extension in the report, and return it if found.
+    pub fn parse_taskprov_advertisement(
+        taskprov_advertisement: &str,
+        task_id: &TaskId,
+        version: DapVersion,
+    ) -> Result<TaskprovAdvertisement, DapAbort> {
+        let taskprov_data = decode_base64url_vec(taskprov_advertisement).ok_or_else(|| {
+            DapAbort::BadRequest(
+                r#"Invalid advertisement in "dap-taskprov" header: base64url parsing failed"#
+                    .to_string(),
+            )
+        })?;
+
+        if compute_task_id(taskprov_data.as_ref()) != *task_id {
+            // Return unrecognizedTask following section 5.1 of the taskprov draft.
+            return Err(DapAbort::UnrecognizedTask { task_id: *task_id });
+        }
+
+        // Return unrecognizedMessage if parsing fails following section 5.1 of the taskprov draft.
+        let task_config =
+            TaskprovAdvertisement::get_decoded_with_param(&version, taskprov_data.as_ref())
+                .map_err(|e| DapAbort::from_codec_error(e, *task_id))?;
+
+        Ok(task_config)
+    }
+
+    pub fn serialize_to_header_value(&self, version: DapVersion) -> Result<String, DapError> {
+        let encoded = self
+            .get_encoded_with_param(&version)
+            .map_err(DapError::encoding)?;
+        Ok(encode_base64url(encoded))
+    }
+
+    #[cfg(any(test, feature = "test-utils"))]
+    pub fn compute_task_id(&self, version: DapVersion) -> TaskId {
+        compute_task_id(&self.get_encoded_with_param(&version).unwrap())
+    }
+}
+
+impl ParameterizedEncode<DapVersion> for TaskprovAdvertisement {
     fn encode_with_param(
         &self,
         version: &DapVersion,
@@ -437,7 +481,7 @@ impl ParameterizedEncode<DapVersion> for TaskConfig {
     }
 }
 
-impl ParameterizedDecode<DapVersion> for TaskConfig {
+impl ParameterizedDecode<DapVersion> for TaskprovAdvertisement {
     fn decode_with_param(
         version: &DapVersion,
         bytes: &mut Cursor<&[u8]>,
@@ -463,7 +507,7 @@ impl ParameterizedDecode<DapVersion> for TaskConfig {
             VdafConfig::decode_with_param(&(version, len), inner)
         })?;
 
-        Ok(TaskConfig {
+        Ok(TaskprovAdvertisement {
             task_info,
             leader_url,
             helper_url,
@@ -481,7 +525,7 @@ mod tests {
     use super::*;
 
     fn read_task_config(version: DapVersion) {
-        let want = TaskConfig {
+        let want = TaskprovAdvertisement {
             task_info: b"this is a cool task!".to_vec(),
             leader_url: UrlBytes {
                 bytes: b"http://exmaple.com/v02".to_vec(),
@@ -512,7 +556,8 @@ mod tests {
             188, 79, 242, 5, 57, 0, 0, 0, 55, 2, 0, 0, 0, 57, 0, 0, 0, 5, 104, 191, 187, 40, 0, 11,
             0, 1, 1, 255, 255, 0, 0, 0, 1, 134, 159,
         ];
-        let got = TaskConfig::get_decoded_with_param(&version, &task_config_bytes).unwrap();
+        let got =
+            TaskprovAdvertisement::get_decoded_with_param(&version, &task_config_bytes).unwrap();
         assert_eq!(got, want);
     }
 
@@ -691,6 +736,43 @@ mod tests {
 
     test_versions! { roundtrip_vdaf_config_pine64_hmac_sha256_aes128 }
 
+    fn roundtrip_taskprov_advertisement(version: DapVersion) {
+        let taskprov_advertisement = TaskprovAdvertisement {
+            task_info: b"this is a cool task!".to_vec(),
+            leader_url: UrlBytes {
+                bytes: b"http://exmaple.com/v02".to_vec(),
+            },
+            helper_url: UrlBytes {
+                bytes: b"https://someservice.cloudflareresearch.com".to_vec(),
+            },
+            query_config: QueryConfig {
+                time_precision: 12_341_234,
+                max_batch_query_count: 1337,
+                min_batch_size: 55,
+                var: QueryConfigVar::FixedSize { max_batch_size: 57 },
+            },
+            task_expiration: 23_232_232_232,
+            vdaf_config: VdafConfig {
+                dp_config: DpConfig::None,
+                var: VdafTypeVar::Prio2 { dimension: 99_999 },
+            },
+        };
+
+        assert_eq!(
+            taskprov_advertisement,
+            TaskprovAdvertisement::parse_taskprov_advertisement(
+                &taskprov_advertisement
+                    .serialize_to_header_value(version)
+                    .unwrap(),
+                &taskprov_advertisement.compute_task_id(version),
+                version,
+            )
+            .unwrap()
+        );
+    }
+
+    test_versions! { roundtrip_taskprov_advertisement }
+
     fn roundtrip_vdaf_config_not_implemented(version: DapVersion) {
         let vdaf_config = VdafConfig {
             dp_config: DpConfig::None,
diff --git a/crates/daphne/src/roles/leader/mod.rs b/crates/daphne/src/roles/leader/mod.rs
index 6941139..8144f2c 100644
--- a/crates/daphne/src/roles/leader/mod.rs
+++ b/crates/daphne/src/roles/leader/mod.rs
@@ -18,9 +18,9 @@ use crate::{
     error::DapAbort,
     fatal_error,
     messages::{
-        request::resource, AggregateShare, AggregateShareReq, AggregationJobId, AggregationJobResp,
-        Base64Encode, BatchId, BatchSelector, Collection, CollectionJobId, CollectionReq, Interval,
-        PartialBatchSelector, Query, Report, TaskId,
+        request::resource, taskprov::TaskprovAdvertisement, AggregateShare, AggregateShareReq,
+        AggregationJobId, AggregationJobResp, Base64Encode, BatchId, BatchSelector, Collection,
+        CollectionJobId, CollectionReq, Interval, PartialBatchSelector, Query, Report, TaskId,
     },
     metrics::{DaphneRequestType, ReportStatus},
     roles::resolve_task_config,
@@ -34,7 +34,7 @@ struct LeaderHttpRequestOptions<'p, P> {
     resp_media_type: DapMediaType,
     req_data: P,
     method: LeaderHttpRequestMethod,
-    taskprov: Option<String>,
+    taskprov_advertisement: Option<TaskprovAdvertisement>,
 }
 
 enum LeaderHttpRequestMethod {
@@ -57,7 +57,7 @@ where
         resp_media_type,
         req_data,
         method,
-        taskprov,
+        taskprov_advertisement,
     } = opts;
 
     let url = task_config
@@ -69,7 +69,7 @@ where
         version: task_config.version,
         media_type: Some(req_media_type),
         task_id: *task_id,
-        taskprov,
+        taskprov_advertisement,
     };
 
     let resp = match method {
@@ -295,7 +295,7 @@ async fn run_agg_job<A: DapLeader>(
 ) -> Result<u64, DapError> {
     let metrics = aggregator.metrics();
 
-    let taskprov = task_config.resolve_taskprove_advertisement()?;
+    let taskprov_advertisement = task_config.resolve_taskprove_advertisement()?;
 
     // Prepare AggregationJobInitReq.
     let agg_job_id = AggregationJobId(thread_rng().gen());
@@ -332,7 +332,7 @@ async fn run_agg_job<A: DapLeader>(
             resp_media_type: DapMediaType::AggregationJobResp,
             req_data: agg_job_init_req,
             method: LeaderHttpRequestMethod::Put,
-            taskprov: taskprov.clone(),
+            taskprov_advertisement: taskprov_advertisement.clone(),
         },
     )
     .await?;
@@ -401,7 +401,7 @@ async fn run_coll_job<A: DapLeader>(
     debug!("collecting id {coll_job_id}");
     let leader_agg_share = aggregator.get_agg_share(task_id, batch_sel).await?;
 
-    let taskprov = task_config.resolve_taskprove_advertisement()?;
+    let taskprov_advertisement = task_config.resolve_taskprove_advertisement()?;
 
     // Check the batch size. If not not ready, then return early.
     //
@@ -445,7 +445,7 @@ async fn run_coll_job<A: DapLeader>(
             resp_media_type: DapMediaType::AggregateShare,
             req_data: agg_share_req,
             method: LeaderHttpRequestMethod::Post,
-            taskprov,
+            taskprov_advertisement,
         },
     )
     .await?;
diff --git a/crates/daphne/src/roles/mod.rs b/crates/daphne/src/roles/mod.rs
index 558bc37..a3b130d 100644
--- a/crates/daphne/src/roles/mod.rs
+++ b/crates/daphne/src/roles/mod.rs
@@ -9,7 +9,8 @@ pub mod leader;
 
 use crate::{
     messages::{Base64Encode, Query, TaskId, Time},
-    taskprov, DapAbort, DapError, DapGlobalConfig, DapQueryConfig, DapRequestMeta, DapTaskConfig,
+    taskprov::DapTaskConfigNeedsOptIn,
+    DapAbort, DapError, DapGlobalConfig, DapQueryConfig, DapRequestMeta, DapTaskConfig,
 };
 
 pub use aggregator::{DapAggregator, DapReportInitializer};
@@ -98,17 +99,19 @@ async fn resolve_task_config(
         config
     } else {
         let taskprov_config = agg.get_taskprov_config();
-        let (Some(taskprov_msg), Some(taskprov_config)) = (&req.taskprov, taskprov_config) else {
+        let (Some(taskprov_advertisement), Some(taskprov_config)) =
+            (&req.taskprov_advertisement, taskprov_config)
+        else {
             return Err(DapAbort::UnrecognizedTask {
                 task_id: req.task_id,
             }
             .into());
         };
-        let task_config = taskprov::resolve_advertised_task_config(
-            taskprov_msg,
+        let task_config = DapTaskConfigNeedsOptIn::try_from_taskprov(
             req.version,
-            taskprov_config,
             &req.task_id,
+            taskprov_advertisement.clone(),
+            taskprov_config,
         )?;
         let task_config = agg.taskprov_opt_in(&req.task_id, task_config).await?;
         agg.taskprov_put(&req.task_id, task_config.clone()).await?;
@@ -1458,7 +1461,7 @@ mod test {
                     version,
                     media_type: Some(DapMediaType::Report),
                     task_id,
-                    taskprov: Some(taskprov_advertisement.clone()),
+                    taskprov_advertisement: Some(taskprov_advertisement.clone()),
                 },
                 resource_id: resource::None,
                 payload: report,
diff --git a/crates/daphne/src/taskprov.rs b/crates/daphne/src/taskprov.rs
index 88fde8e..41f6347 100644
--- a/crates/daphne/src/taskprov.rs
+++ b/crates/daphne/src/taskprov.rs
@@ -12,8 +12,8 @@ use crate::{
     fatal_error,
     hpke::HpkeConfig,
     messages::{
-        self, decode_base64url_vec,
-        taskprov::{QueryConfigVar, TaskConfig, VdafTypeVar},
+        self,
+        taskprov::{QueryConfigVar, TaskprovAdvertisement, VdafTypeVar},
         Duration, TaskId, Time,
     },
     roles::aggregator::TaskprovConfig,
@@ -27,7 +27,6 @@ use crate::{
         Prio3Config,
     },
 };
-use prio::codec::ParameterizedDecode;
 use ring::{
     digest,
     hkdf::{Prk, Salt, HKDF_SHA256},
@@ -105,50 +104,6 @@ fn malformed_task_config(task_id: &TaskId, detail: String) -> DapAbort {
     }
 }
 
-/// Convert a task config advertised by the peer into a [`DapTaskConfig`].
-///
-/// 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(
-    taskprov_msg: &str,
-    version: DapVersion,
-    taskprov_config: TaskprovConfig<'_>,
-    task_id: &TaskId,
-) -> Result<DapTaskConfigNeedsOptIn, DapAbort> {
-    let task_config = get_taskprov_task_config(taskprov_msg, task_id, version)?;
-    DapTaskConfigNeedsOptIn::try_from_taskprov(
-        version,
-        task_id, // get_taskprov_task_config() checks that this matches the derived ID
-        task_config,
-        taskprov_config,
-    )
-}
-
-/// Check for a taskprov extension in the report, and return it if found.
-fn get_taskprov_task_config(
-    taskprov_msg: &str,
-    task_id: &TaskId,
-    version: DapVersion,
-) -> Result<TaskConfig, DapAbort> {
-    let taskprov_data = decode_base64url_vec(taskprov_msg).ok_or_else(|| {
-        DapAbort::BadRequest(
-            r#"Invalid advertisement in "dap-taskprov" header: base64url parsing failed"#
-                .to_string(),
-        )
-    })?;
-
-    if compute_task_id(taskprov_data.as_ref()) != *task_id {
-        // Return unrecognizedTask following section 5.1 of the taskprov draft.
-        return Err(DapAbort::UnrecognizedTask { task_id: *task_id });
-    }
-
-    // Return unrecognizedMessage if parsing fails following section 5.1 of the taskprov draft.
-    let task_config = TaskConfig::get_decoded_with_param(&version, taskprov_data.as_ref())
-        .map_err(|e| DapAbort::from_codec_error(e, *task_id))?;
-
-    Ok(task_config)
-}
-
 fn url_from_bytes(task_id: &TaskId, url_bytes: &[u8]) -> Result<Url, DapAbort> {
     let url_string = std::str::from_utf8(url_bytes).map_err(|e| {
         malformed_task_config(
@@ -338,7 +293,7 @@ impl DapTaskConfigNeedsOptIn {
     pub(crate) fn try_from_taskprov(
         version: DapVersion,
         task_id: &TaskId,
-        task_config: TaskConfig,
+        task_config: TaskprovAdvertisement,
         taskprov_config: TaskprovConfig<'_>,
     ) -> Result<Self, DapAbort> {
         // Only one query per batch is currently supported.
@@ -458,7 +413,7 @@ impl TryFrom<&VdafConfig> for messages::taskprov::VdafTypeVar {
     }
 }
 
-impl TryFrom<&DapTaskConfig> for messages::taskprov::TaskConfig {
+impl TryFrom<&DapTaskConfig> for messages::taskprov::TaskprovAdvertisement {
     type Error = DapError;
 
     fn try_from(task_config: &DapTaskConfig) -> Result<Self, DapError> {
@@ -497,11 +452,11 @@ mod test {
 
     use prio::codec::ParameterizedEncode;
 
-    use super::{compute_task_id, compute_vdaf_verify_key, resolve_advertised_task_config};
+    use super::{compute_task_id, compute_vdaf_verify_key};
     use crate::{
         error::DapAbort,
         hpke::{HpkeKemId, HpkeReceiverConfig},
-        messages::{self, encode_base64url, TaskId},
+        messages::{self, TaskId},
         taskprov::{DapTaskConfigNeedsOptIn, OptInParam},
         test_versions,
         vdaf::{VdafConfig, VdafVerifyKey},
@@ -510,7 +465,7 @@ mod test {
 
     /// Test conversion between the serialized task configuration and a `DapTaskConfig`.
     fn try_from_taskprov(version: DapVersion) {
-        let taskprov_config = messages::taskprov::TaskConfig {
+        let taskprov_config = messages::taskprov::TaskprovAdvertisement {
             task_info: "cool task".as_bytes().to_vec(),
             leader_url: messages::taskprov::UrlBytes {
                 bytes: b"https://leader.com/".to_vec(),
@@ -551,7 +506,7 @@ mod test {
         });
 
         assert_eq!(
-            messages::taskprov::TaskConfig::try_from(&task_config).unwrap(),
+            messages::taskprov::TaskprovAdvertisement::try_from(&task_config).unwrap(),
             taskprov_config
         );
     }
@@ -590,7 +545,7 @@ mod test {
     fn resolve_advertised_task_config_expect_abort_unrecognized_vdaf(version: DapVersion) {
         // Create a request for a taskprov task with an unrecognized VDAF.
         let (req, task_id) = {
-            let taskprov_task_config_bytes = messages::taskprov::TaskConfig {
+            let taskprov_advertisement = messages::taskprov::TaskprovAdvertisement {
                 task_info: "cool task".as_bytes().to_vec(),
                 leader_url: messages::taskprov::UrlBytes {
                     bytes: b"https://leader.com/".to_vec(),
@@ -613,16 +568,19 @@ mod test {
                         param: b"vdaf type param".to_vec(),
                     },
                 },
-            }
-            .get_encoded_with_param(&version)
-            .unwrap();
-            let task_id = compute_task_id(&taskprov_task_config_bytes);
-            let taskprov_task_config_base64url = encode_base64url(&taskprov_task_config_bytes);
+            };
+            let task_id = {
+                compute_task_id(
+                    &taskprov_advertisement
+                        .get_encoded_with_param(&version)
+                        .unwrap(),
+                )
+            };
 
             let req = DapRequestMeta {
                 version,
                 task_id,
-                taskprov: Some(taskprov_task_config_base64url),
+                taskprov_advertisement: Some(taskprov_advertisement),
                 media_type: None, // ignored by test
             };
 
@@ -634,14 +592,14 @@ mod test {
             .config;
 
         assert_matches::assert_matches!(
-            resolve_advertised_task_config(
-                req.taskprov.as_deref().unwrap(),
+            DapTaskConfigNeedsOptIn::try_from_taskprov(
                 req.version,
+                &task_id,
+                req.taskprov_advertisement.unwrap(),
                 crate::roles::aggregator::TaskprovConfig {
                     vdaf_verify_key_init: &[0; 32],
                     hpke_collector_config: &collector_hpke_config,
                 },
-                &task_id
             ).unwrap_err(),
             DapAbort::InvalidTask{ detail, .. } if detail == "unimplemented VDAF type (1337)"
         );
@@ -652,7 +610,7 @@ mod test {
     fn resolve_advertised_task_config_ignore_unimplemented_dp_ocnfig(version: DapVersion) {
         // Create a request for a taskprov task with an unrecognized DP mechanism.
         let (req, task_id) = {
-            let taskprov_task_config_bytes = messages::taskprov::TaskConfig {
+            let taskprov_advertisement = messages::taskprov::TaskprovAdvertisement {
                 task_info: "cool task".as_bytes().to_vec(),
                 leader_url: messages::taskprov::UrlBytes {
                     bytes: b"https://leader.com/".to_vec(),
@@ -674,16 +632,19 @@ mod test {
                     },
                     var: messages::taskprov::VdafTypeVar::Prio2 { dimension: 1337 },
                 },
-            }
-            .get_encoded_with_param(&version)
-            .unwrap();
-            let task_id = compute_task_id(&taskprov_task_config_bytes);
-            let taskprov_task_config_base64url = encode_base64url(&taskprov_task_config_bytes);
+            };
+            let task_id = {
+                compute_task_id(
+                    &taskprov_advertisement
+                        .get_encoded_with_param(&version)
+                        .unwrap(),
+                )
+            };
 
             let req = crate::DapRequestMeta {
                 version,
                 task_id,
-                taskprov: Some(taskprov_task_config_base64url),
+                taskprov_advertisement: Some(taskprov_advertisement),
                 media_type: None,
             };
 
@@ -694,14 +655,14 @@ mod test {
             .unwrap()
             .config;
 
-        let _ = resolve_advertised_task_config(
-            req.taskprov.as_deref().unwrap(),
+        let _ = DapTaskConfigNeedsOptIn::try_from_taskprov(
             req.version,
+            &task_id,
+            req.taskprov_advertisement.unwrap(),
             crate::roles::aggregator::TaskprovConfig {
                 hpke_collector_config: &collector_hpke_config,
                 vdaf_verify_key_init: &[0; 32],
             },
-            &task_id,
         )
         .unwrap();
     }


Test patch:
diff --git a/crates/dapf/src/acceptance/load_testing.rs b/crates/dapf/src/acceptance/load_testing.rs
index d0266c6..2b88de2 100644
--- a/crates/dapf/src/acceptance/load_testing.rs
+++ b/crates/dapf/src/acceptance/load_testing.rs
@@ -462,7 +462,7 @@ pub async fn execute_single_combination_from_env(
                     leader_agg_share,
                     batch_id,
                     VERSION,
-                    test_task_config.taskprov_advertisement.as_deref(),
+                    test_task_config.taskprov_advertisement.as_ref(),
                     test_task_config.task_id,
                 )
                 .await
diff --git a/crates/daphne-server/tests/e2e/test_runner.rs b/crates/daphne-server/tests/e2e/test_runner.rs
index abae06d..1db84fb 100644
--- a/crates/daphne-server/tests/e2e/test_runner.rs
+++ b/crates/daphne-server/tests/e2e/test_runner.rs
@@ -7,8 +7,8 @@ use daphne::{
     constants::DapMediaType,
     hpke::{HpkeAeadId, HpkeConfig, HpkeKdfId, HpkeKemId, HpkeReceiverConfig},
     messages::{
-        encode_base64url, Base64Encode, BatchId, CollectionJobId, Duration, HpkeConfigList,
-        Interval, TaskId,
+        encode_base64url, taskprov::TaskprovAdvertisement, Base64Encode, BatchId, CollectionJobId,
+        Duration, HpkeConfigList, Interval, TaskId,
     },
     vdaf::{Prio3Config, VdafConfig},
     DapGlobalConfig, DapLeaderProcessTelemetry, DapQueryConfig, DapTaskConfig, DapVersion,
@@ -473,7 +473,7 @@ impl TestRunner {
         client: &reqwest::Client,
         path: &str,
         media_type: DapMediaType,
-        taskprov: Option<&str>,
+        taskprov: Option<&TaskprovAdvertisement>,
         data: Vec<u8>,
     ) -> anyhow::Result<()> {
         let url = self.leader_url.join(path)?;
@@ -492,7 +492,9 @@ impl TestRunner {
         if let Some(taskprov_advertisement) = taskprov {
             headers.insert(
                 reqwest::header::HeaderName::from_static(http_headers::DAP_TASKPROV),
-                reqwest::header::HeaderValue::from_str(taskprov_advertisement)?,
+                reqwest::header::HeaderValue::from_str(
+                    &taskprov_advertisement.serialize_to_header_value(self.version)?,
+                )?,
             );
         }
 
@@ -579,7 +581,7 @@ impl TestRunner {
         &self,
         client: &reqwest::Client,
         token: &str,
-        taskprov: Option<&str>,
+        taskprov: Option<&TaskprovAdvertisement>,
         task_id: Option<&TaskId>,
         data: Vec<u8>,
     ) -> anyhow::Result<Url> {
@@ -601,7 +603,9 @@ impl TestRunner {
         if let Some(taskprov_advertisement) = taskprov {
             headers.insert(
                 reqwest::header::HeaderName::from_static(http_headers::DAP_TASKPROV),
-                reqwest::header::HeaderValue::from_str(taskprov_advertisement)?,
+                reqwest::header::HeaderValue::from_str(
+                    &taskprov_advertisement.serialize_to_header_value(self.version)?,
+                )?,
             );
         }