Repository: cloudflare/daphne
Pull request #768 (2025-01-24 15:25:28)

DAP-13: Reject reports with timestamps before task start time
Partially addresses #698.
Replaces #755.

Related changes:

* Adjust the bound checks slightly: we should accept a timestamp this
  equal to either the start time or end time.

* Check the report validity window (for replay protection) before task
  validity. This isn't strictly necessary, but it simplifies some unit
  tests.

* Update the representation of the `InternalTestAddTask` struct to
  include the task duration in DAP-13. Note that the format of this
  struct is specified by draft-dcook-ppm-dap-interop-test-design, which
  has not yet been updated for DAP-13. We may need to revise the format
  at a later date.

* Update the report interval computation in the end-to-end tests to
  avoid using a timestamp that precedes the task start time.

* Add more tracing for debugging.


Patch:
diff --git a/crates/dapf/src/main.rs b/crates/dapf/src/main.rs
index 685285a..6df408c 100644
--- a/crates/dapf/src/main.rs
+++ b/crates/dapf/src/main.rs
@@ -19,7 +19,7 @@ use daphne::{
     messages::{
         self, encode_base64url, BatchSelector, CollectionReq, PartialBatchSelector, Query, TaskId,
     },
-    DapBatchMode, DapMeasurement, DapVersion,
+    DapBatchMode, DapMeasurement, DapTaskLifetime, DapVersion,
 };
 use daphne_service_utils::{
     bearer_token::BearerToken,
@@ -837,7 +837,7 @@ async fn handle_test_routes(action: TestAction, http_client: HttpClient) -> anyh
             min_batch_size,
             collector_hpke_config,
             time_precision,
-            expires_in_seconds: task_expiration,
+            expires_in_seconds,
         } => {
             let vdaf = use_or_request_from_user_or_default(vdaf, CliVdafConfig::default, "vdaf")?
                 .into_vdaf_config();
@@ -850,7 +850,40 @@ async fn handle_test_routes(action: TestAction, http_client: HttpClient) -> anyh
                 "query",
             )?;
             let role = use_or_request_from_user(role, "role")?;
+
+            let leader = use_or_request_from_user(leader_url, "leader url")?;
+            let helper = use_or_request_from_user(helper_url, "helper url")?;
+
             let default_task_id = TaskId(thread_rng().gen());
+
+            let version = match role {
+                DapAggregatorRole::Leader => deduce_dap_version_from_url(&leader)?,
+                DapAggregatorRole::Helper => deduce_dap_version_from_url(&helper)?,
+            };
+
+            let lifetime = {
+                let not_before = SystemTime::now()
+                    .duration_since(SystemTime::UNIX_EPOCH)
+                    .unwrap()
+                    .as_secs();
+
+                let task_duration = use_or_request_from_user_or_default(
+                    expires_in_seconds,
+                    || 604_800u64,
+                    "task should expire in",
+                )?;
+
+                match version {
+                    DapVersion::Latest => DapTaskLifetime::Latest {
+                        start: not_before,
+                        duration: task_duration,
+                    },
+                    DapVersion::Draft09 => DapTaskLifetime::Draft09 {
+                        expiration: not_before + task_duration,
+                    },
+                }
+            };
+
             let internal_task = InternalTestAddTask {
                 task_id: use_or_request_from_user_or_default(
                     task_id,
@@ -858,8 +891,8 @@ async fn handle_test_routes(action: TestAction, http_client: HttpClient) -> anyh
                     "task id",
                 )?
                 .into(),
-                leader: use_or_request_from_user(leader_url, "leader url")?,
-                helper: use_or_request_from_user(helper_url, "helper url")?,
+                leader,
+                helper,
                 vdaf: InternalTestVdaf::from(vdaf),
                 leader_authentication_token: use_or_request_from_user(
                     leader_auth_token,
@@ -898,15 +931,7 @@ async fn handle_test_routes(action: TestAction, http_client: HttpClient) -> anyh
                     collector_hpke_config,
                     "collector hpke config",
                 )?,
-                task_expiration: SystemTime::now()
-                    .duration_since(SystemTime::UNIX_EPOCH)
-                    .unwrap()
-                    .as_secs()
-                    + use_or_request_from_user_or_default(
-                        task_expiration,
-                        || 604_800u64,
-                        "task should expire in",
-                    )?,
+                lifetime,
             };
 
             print_json(&internal_task);
diff --git a/crates/daphne-server/src/roles/mod.rs b/crates/daphne-server/src/roles/mod.rs
index 0e27339..4b8a534 100644
--- a/crates/daphne-server/src/roles/mod.rs
+++ b/crates/daphne-server/src/roles/mod.rs
@@ -113,7 +113,7 @@ mod test_utils {
         messages::decode_base64url_vec,
         roles::DapAggregator,
         vdaf::{Prio3Config, VdafConfig},
-        DapBatchMode, DapError, DapTaskConfig, DapVersion,
+        DapBatchMode, DapError, DapTaskConfig, DapTaskLifetime, DapVersion,
     };
     use daphne_service_utils::{
         bearer_token::BearerToken,
@@ -297,6 +297,19 @@ mod test_utils {
                 }
             };
 
+            let (not_before, not_after) = match cmd.lifetime {
+                DapTaskLifetime::Latest { start, duration } => {
+                    (start, start.saturating_add(duration))
+                }
+                // draft09 compatibility: Previously the task start time was not conveyed by the
+                // taskprov advertisement, so just use the current time.
+                DapTaskLifetime::Draft09 { expiration } => (self.get_current_time(), expiration),
+            };
+
+            if not_before >= not_after {
+                tracing::warn!("task has empty validity interval: not_before={not_before}; not_after={not_after}");
+            }
+
             if self
                 .kv()
                 .put_if_not_exists_with_expiration::<kv::prefix::TaskConfig>(
@@ -306,8 +319,8 @@ mod test_utils {
                         leader_url: cmd.leader,
                         helper_url: cmd.helper,
                         time_precision: cmd.time_precision,
-                        not_before: self.get_current_time(),
-                        not_after: cmd.task_expiration,
+                        not_before,
+                        not_after,
                         min_batch_size: cmd.min_batch_size,
                         query,
                         vdaf,
@@ -316,7 +329,7 @@ mod test_utils {
                         method: Default::default(),
                         num_agg_span_shards: NonZeroUsize::new(4).unwrap(),
                     },
-                    cmd.task_expiration,
+                    not_after,
                 )
                 .await
                 .map_err(|e| fatal_error!(err = ?e, "failed to put task config in kv"))?
diff --git a/crates/daphne-server/src/router/extractor.rs b/crates/daphne-server/src/router/extractor.rs
index ef088c1..51d0896 100644
--- a/crates/daphne-server/src/router/extractor.rs
+++ b/crates/daphne-server/src/router/extractor.rs
@@ -675,10 +675,8 @@ mod test {
             min_batch_size: 1,
             query_config: daphne::messages::taskprov::QueryConfig::TimeInterval,
             lifetime: match version {
-                DapVersion::Draft09 => {
-                    daphne::messages::taskprov::TaskLifetime::Draft09 { expiration: 1 }
-                }
-                DapVersion::Latest => daphne::messages::taskprov::TaskLifetime::Latest {
+                DapVersion::Draft09 => daphne::DapTaskLifetime::Draft09 { expiration: 1 },
+                DapVersion::Latest => daphne::DapTaskLifetime::Latest {
                     start: 0,
                     duration: 1,
                 },
diff --git a/crates/daphne-service-utils/src/compute_offload/compute_offload.capnp b/crates/daphne-service-utils/src/compute_offload/compute_offload.capnp
index 19f1858..581baf0 100644
--- a/crates/daphne-service-utils/src/compute_offload/compute_offload.capnp
+++ b/crates/daphne-service-utils/src/compute_offload/compute_offload.capnp
@@ -44,13 +44,14 @@ struct HpkeReceiverConfig @0xeec9b4a50458edb7 {
     privateKey @1 :Data;
 }
 
-struct PartialDapTaskConfig @0xdcc9bf18fc62d406 {
+struct PartialDapTaskConfig @0xb11c76132b15968a {
 
     version             @0  :Base.DapVersion;
     methodIsTaskprov    @1  :Bool;
-    notAfter            @2  :Base.Time;
-    vdaf                @3  :VdafConfig;
-    vdafVerifyKey       @4  :VdafVerifyKey;
+    notBefore           @2  :Base.Time;
+    notAfter            @3  :Base.Time;
+    vdaf                @4  :VdafConfig;
+    vdafVerifyKey       @5  :VdafVerifyKey;
 }
 
 struct PublicExtensionsList @0x8b3c98c0ddd0043e {
diff --git a/crates/daphne-service-utils/src/compute_offload/mod.rs b/crates/daphne-service-utils/src/compute_offload/mod.rs
index 6a14891..142c1d9 100644
--- a/crates/daphne-service-utils/src/compute_offload/mod.rs
+++ b/crates/daphne-service-utils/src/compute_offload/mod.rs
@@ -203,12 +203,14 @@ impl CapnprotoPayloadEncode for PartialDapTaskConfigForReportInit<'_> {
 
     fn encode_to_builder(&self, mut builder: Self::Builder<'_>) {
         let PartialDapTaskConfigForReportInit {
+            not_before,
             not_after,
             method_is_taskprov,
             version,
             vdaf,
             vdaf_verify_key,
         } = self;
+        builder.set_not_before(*not_before);
         builder.set_not_after(*not_after);
         builder.set_method_is_taskprov(*method_is_taskprov);
         builder.set_version((*version).into());
@@ -227,6 +229,7 @@ impl CapnprotoPayloadDecode for PartialDapTaskConfigForReportInit<'static> {
         Self: Sized,
     {
         Ok(Self {
+            not_before: reader.get_not_before(),
             not_after: reader.get_not_after(),
             method_is_taskprov: reader.get_method_is_taskprov(),
             version: reader.get_version()?.into(),
diff --git a/crates/daphne-worker/src/aggregator/roles/mod.rs b/crates/daphne-worker/src/aggregator/roles/mod.rs
index 5fd7d3d..9f2a8ce 100644
--- a/crates/daphne-worker/src/aggregator/roles/mod.rs
+++ b/crates/daphne-worker/src/aggregator/roles/mod.rs
@@ -114,7 +114,7 @@ mod test_utils {
         messages::decode_base64url_vec,
         roles::DapAggregator as _,
         vdaf::{Prio3Config, VdafConfig},
-        DapBatchMode, DapError, DapTaskConfig, DapVersion,
+        DapBatchMode, DapError, DapTaskConfig, DapTaskLifetime, DapVersion,
     };
     use daphne_service_utils::{
         bearer_token::BearerToken,
@@ -279,6 +279,19 @@ mod test_utils {
                 }
             };
 
+            let (not_before, not_after) = match cmd.lifetime {
+                DapTaskLifetime::Latest { start, duration } => {
+                    (start, start.saturating_add(duration))
+                }
+                // draft09 compatibility: Previously the task start time was not conveyed by the
+                // taskprov advertisement, so just use the current time.
+                DapTaskLifetime::Draft09 { expiration } => (self.get_current_time(), expiration),
+            };
+
+            if not_before >= not_after {
+                tracing::warn!("task has empty validity interval: not_before={not_before}; not_after={not_after}");
+            }
+
             if self
                 .kv()
                 .put_if_not_exists_with_expiration::<kv::prefix::TaskConfig>(
@@ -288,8 +301,8 @@ mod test_utils {
                         leader_url: cmd.leader,
                         helper_url: cmd.helper,
                         time_precision: cmd.time_precision,
-                        not_before: self.get_current_time(),
-                        not_after: cmd.task_expiration,
+                        not_before,
+                        not_after,
                         min_batch_size: cmd.min_batch_size,
                         query,
                         vdaf,
@@ -298,7 +311,7 @@ mod test_utils {
                         method: Default::default(),
                         num_agg_span_shards: NonZeroUsize::new(4).unwrap(),
                     },
-                    cmd.task_expiration,
+                    not_after,
                 )
                 .await
                 .map_err(|e| fatal_error!(err = ?e, "failed to put task config in kv"))?
diff --git a/crates/daphne-worker/src/aggregator/router/extractor.rs b/crates/daphne-worker/src/aggregator/router/extractor.rs
index dbf0cac..9ea29fe 100644
--- a/crates/daphne-worker/src/aggregator/router/extractor.rs
+++ b/crates/daphne-worker/src/aggregator/router/extractor.rs
@@ -731,10 +731,8 @@ mod test {
             min_batch_size: 1,
             query_config: daphne::messages::taskprov::QueryConfig::TimeInterval,
             lifetime: match version {
-                DapVersion::Draft09 => {
-                    daphne::messages::taskprov::TaskLifetime::Draft09 { expiration: 1 }
-                }
-                DapVersion::Latest => daphne::messages::taskprov::TaskLifetime::Latest {
+                DapVersion::Draft09 => daphne::DapTaskLifetime::Draft09 { expiration: 1 },
+                DapVersion::Latest => daphne::DapTaskLifetime::Latest {
                     start: 0,
                     duration: 1,
                 },
diff --git a/crates/daphne/src/lib.rs b/crates/daphne/src/lib.rs
index 7c91e31..2dce27a 100644
--- a/crates/daphne/src/lib.rs
+++ b/crates/daphne/src/lib.rs
@@ -105,7 +105,7 @@ pub enum DapVersion {
     #[default]
     Draft09,
 
-    #[serde(rename = "v10")]
+    #[serde(rename = "v13")]
     Latest,
 }
 
@@ -114,7 +114,7 @@ impl FromStr for DapVersion {
     fn from_str(version: &str) -> Result<Self, Self::Err> {
         match version {
             "v09" => Ok(DapVersion::Draft09),
-            "v10" => Ok(DapVersion::Latest),
+            "v13" => Ok(DapVersion::Latest),
             _ => Err(DapAbort::version_unknown()),
         }
     }
@@ -124,7 +124,7 @@ impl AsRef<str> for DapVersion {
     fn as_ref(&self) -> &str {
         match self {
             DapVersion::Draft09 => "v09",
-            DapVersion::Latest => "v10",
+            DapVersion::Latest => "v13",
         }
     }
 }
@@ -141,7 +141,7 @@ impl std::fmt::Display for DapVersion {
 pub struct DapGlobalConfig {
     /// Maximum interval duration permitted in [`CollectionReq`](messages::CollectionReq).
     ///
-    /// Prevents Collectors from requesting wide range or reports.
+    /// Prevents Collectors from requesting a wide range of reports.
     pub max_batch_duration: Duration,
 
     /// Lower bound of an acceptable batch interval for collect requests.
@@ -568,11 +568,7 @@ impl DapTaskParameters {
             time_precision: self.time_precision,
             min_batch_size: self.min_batch_size.try_into().unwrap(),
             query_config: (&self.query).try_into()?,
-            lifetime: messages::taskprov::TaskLifetime::from_validity_range(
-                self.version,
-                not_before,
-                not_after,
-            ),
+            lifetime: DapTaskLifetime::from_validity_range(self.version, not_before, not_after),
             vdaf_config: (&self.vdaf).try_into()?,
             extensions: Vec::new(),
             draft09_max_batch_query_count: match self.version {
@@ -784,6 +780,13 @@ impl DapTaskConfig {
     pub fn method_is_taskprov(&self) -> bool {
         matches!(self.method, DapTaskConfigMethod::Taskprov { .. })
     }
+
+    /// Return the task lifetime.
+    //
+    // draft09 compatibility: The information conveyed by the lifetime depends on the DAP version.
+    pub fn lifetime(&self) -> DapTaskLifetime {
+        DapTaskLifetime::from_validity_range(self.version, self.not_before, self.not_after)
+    }
 }
 
 impl AsRef<DapTaskConfig> for DapTaskConfig {
@@ -1070,3 +1073,35 @@ pub struct DapLeaderProcessTelemetry {
     /// The number of reports processed.
     pub reports_processed: u64,
 }
+
+/// Task lifetime parameters.
+#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
+pub enum DapTaskLifetime {
+    Latest {
+        /// Task start time.
+        start: Time,
+        /// Task duration.
+        duration: Duration,
+    },
+    /// draft09 compatibility: Previously the DAP task parameters (and thus Taskprov) only
+    /// expressed an end time and not a start time.
+    Draft09 { expiration: Time },
+}
+
+impl DapTaskLifetime {
+    pub(crate) fn from_validity_range(
+        version: DapVersion,
+        not_before: Time,
+        not_after: Time,
+    ) -> Self {
+        match version {
+            DapVersion::Draft09 => Self::Draft09 {
+                expiration: not_after,
+            },
+            DapVersion::Latest => Self::Latest {
+                start: not_before,
+                duration: not_after - not_before,
+            },
+        }
+    }
+}
diff --git a/crates/daphne/src/messages/taskprov.rs b/crates/daphne/src/messages/taskprov.rs
index ebeb184..ddcf1d5 100644
--- a/crates/daphne/src/messages/taskprov.rs
+++ b/crates/daphne/src/messages/taskprov.rs
@@ -3,12 +3,16 @@
 
 //! draft-wang-ppm-dap-taskprov: Messages for the taskrpov extension for DAP.
 
-use crate::messages::{
-    decode_u16_bytes, encode_u16_bytes, Duration, Time, BATCH_MODE_LEADER_SELECTED,
-    BATCH_MODE_TIME_INTERVAL,
+use crate::{
+    error::DapAbort,
+    messages::{
+        decode_u16_bytes, encode_u16_bytes, Duration, Time, BATCH_MODE_LEADER_SELECTED,
+        BATCH_MODE_TIME_INTERVAL,
+    },
+    pine::PineParam,
+    taskprov::compute_task_id,
+    DapError, DapTaskLifetime, DapVersion,
 };
-use crate::pine::PineParam;
-use crate::{DapError, DapVersion};
 use prio::codec::{
     decode_u16_items, decode_u8_items, encode_u16_items, encode_u8_items, CodecError, Decode,
     Encode, ParameterizedDecode, ParameterizedEncode,
@@ -20,8 +24,6 @@ use std::num::NonZeroU32;
 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;
@@ -467,38 +469,6 @@ impl ParameterizedDecode<(DapVersion, Option<usize>)> for QueryConfig {
     }
 }
 
-/// Task lifetime parameters.
-#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
-pub enum TaskLifetime {
-    Latest {
-        /// Task start time.
-        start: Time,
-        /// Task duration.
-        duration: Duration,
-    },
-    /// draft09 compatibility: Previously the DAP task parameters (and thus Taskprov) only
-    /// expressed an end time and not a start time.
-    Draft09 { expiration: Time },
-}
-
-impl TaskLifetime {
-    pub(crate) fn from_validity_range(
-        version: DapVersion,
-        not_before: Time,
-        not_after: Time,
-    ) -> Self {
-        match version {
-            DapVersion::Draft09 => Self::Draft09 {
-                expiration: not_after,
-            },
-            DapVersion::Latest => Self::Latest {
-                start: not_before,
-                duration: not_after - not_before,
-            },
-        }
-    }
-}
-
 /// Taskprov extensions.
 #[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
 pub enum TaskprovExtension {
@@ -535,7 +505,7 @@ pub struct TaskprovAdvertisement {
     pub time_precision: Duration,
     pub min_batch_size: u32,
     pub query_config: QueryConfig,
-    pub lifetime: TaskLifetime,
+    pub lifetime: DapTaskLifetime,
     pub vdaf_config: VdafConfig,
     pub extensions: Vec<TaskprovExtension>,
     pub draft09_max_batch_query_count: Option<u16>,
@@ -605,7 +575,7 @@ impl ParameterizedEncode<DapVersion> for TaskprovAdvertisement {
                     self.query_config.encode_with_param(&version, bytes)?;
                     Ok(())
                 })?;
-                let TaskLifetime::Draft09 { expiration } = self.lifetime else {
+                let DapTaskLifetime::Draft09 { expiration } = self.lifetime else {
                     return Err(CodecError::Other("task expiration should be set".into()));
                 };
                 expiration.encode(bytes)?;
@@ -631,7 +601,7 @@ impl ParameterizedEncode<DapVersion> for TaskprovAdvertisement {
                 self.time_precision.encode(bytes)?;
                 self.min_batch_size.encode(bytes)?;
                 self.query_config.encode_with_param(version, bytes)?;
-                let TaskLifetime::Latest { start, duration } = self.lifetime else {
+                let DapTaskLifetime::Latest { start, duration } = self.lifetime else {
                     return Err(CodecError::Other(
                         "task start time and duration should be set".into(),
                     ));
@@ -681,7 +651,7 @@ impl ParameterizedDecode<DapVersion> for TaskprovAdvertisement {
                             query_config,
                         ))
                     })?;
-                let lifetime = TaskLifetime::Draft09 {
+                let lifetime = DapTaskLifetime::Draft09 {
                     expiration: Time::decode(bytes)?,
                 };
                 let (draft09_dp_config, vdaf_config) =
@@ -719,7 +689,7 @@ impl ParameterizedDecode<DapVersion> for TaskprovAdvertisement {
                 let time_precision = Duration::decode(bytes)?;
                 let min_batch_size = u32::decode(bytes)?;
                 let query_config = QueryConfig::decode_with_param(&(*version, None), bytes)?;
-                let lifetime = TaskLifetime::Latest {
+                let lifetime = DapTaskLifetime::Latest {
                     start: Time::decode(bytes)?,
                     duration: Duration::decode(bytes)?,
                 };
@@ -767,7 +737,7 @@ mod tests {
                     DapVersion::Latest => None,
                 },
             },
-            lifetime: TaskLifetime::from_validity_range(version, 23_232_232_232, 23_232_232_232),
+            lifetime: DapTaskLifetime::from_validity_range(version, 23_232_232_232, 23_232_232_232),
             vdaf_config: VdafConfig::Prio2 { dimension: 99_999 },
             extensions: match version {
                 DapVersion::Latest => vec![
@@ -992,7 +962,7 @@ mod tests {
                     DapVersion::Latest => None,
                 },
             },
-            lifetime: TaskLifetime::from_validity_range(version, 23_232_232_232, 23_232_232_232),
+            lifetime: DapTaskLifetime::from_validity_range(version, 23_232_232_232, 23_232_232_232),
             vdaf_config: VdafConfig::Prio2 { dimension: 99_999 },
             extensions: Vec::new(),
             draft09_max_batch_query_count: match version {
@@ -1053,7 +1023,7 @@ mod tests {
                     DapVersion::Latest => None,
                 },
             },
-            lifetime: TaskLifetime::from_validity_range(version, 23_232_232_232, 23_232_232_232),
+            lifetime: DapTaskLifetime::from_validity_range(version, 23_232_232_232, 23_232_232_232),
             vdaf_config: VdafConfig::Prio2 { dimension: 99_999 },
             extensions: Vec::new(),
             draft09_max_batch_query_count: match version {
diff --git a/crates/daphne/src/protocol/aggregator.rs b/crates/daphne/src/protocol/aggregator.rs
index b169f36..ce40ef8 100644
--- a/crates/daphne/src/protocol/aggregator.rs
+++ b/crates/daphne/src/protocol/aggregator.rs
@@ -176,12 +176,14 @@ impl DapTaskConfig {
                     ..
                 } => {
                     // Skip report that can't be processed any further.
+                    tracing::warn!("produce_agg_job_req: rejected report: {failure:?}");
                     metrics.report_inc_by(ReportStatus::Rejected(failure), 1);
                     continue;
                 }
             }
         }
 
+        tracing::debug!("produce_agg_job_req: reports count: {}", prep_inits.len());
         Ok((
             DapAggregationJobState {
                 seq: states,
@@ -252,6 +254,10 @@ impl DapTaskConfig {
                 task_id: *task_id,
             })?;
         }
+        tracing::debug!(
+            "consume_agg_job_req: reports count: {}",
+            agg_job_init_req.prep_inits.len()
+        );
 
         agg_job_init_req
             .prep_inits
@@ -413,6 +419,7 @@ impl DapTaskConfig {
 
                 // Skip report that can't be processed any further.
                 TransitionVar::Failed(err) => {
+                    tracing::warn!("report rejected by Helper: {err:?}");
                     metrics.report_inc_by(ReportStatus::Rejected(*err), 1);
                     continue;
                 }
diff --git a/crates/daphne/src/protocol/mod.rs b/crates/daphne/src/protocol/mod.rs
index 2f39c3e..0da6683 100644
--- a/crates/daphne/src/protocol/mod.rs
+++ b/crates/daphne/src/protocol/mod.rs
@@ -401,12 +401,15 @@ mod test {
 
         let agg_job_init_req = {
             // Temporarily overwrite the valid report time range so that the Leader accepts the
-            // out-of-range report and produces the request.
-            let tmp = t.valid_report_range.clone();
+            // out-of-range report and produces the request. Likewise for the task start time.
+            let tmp_valid_time_range = t.valid_report_range.clone();
+            let tmp_not_before = t.task_config.not_before;
             t.valid_report_range = 0..u64::MAX;
+            t.task_config.not_before = 0;
             let (_, agg_job_init_req) =
                 t.produce_agg_job_req(&DapAggregationParam::Empty, reports.clone());
-            t.valid_report_range = tmp;
+            t.valid_report_range = tmp_valid_time_range;
+            t.task_config.not_before = tmp_not_before;
             agg_job_init_req
         };
         let (_agg_span, agg_job_resp) = t.handle_agg_job_req(agg_job_init_req);
diff --git a/crates/daphne/src/protocol/report_init.rs b/crates/daphne/src/protocol/report_init.rs
index 7502ab5..0cfcd67 100644
--- a/crates/daphne/src/protocol/report_init.rs
+++ b/crates/daphne/src/protocol/report_init.rs
@@ -100,6 +100,7 @@ impl InitializedReport<WithPeerPrepShare> {
 impl<'s> From<&'s DapTaskConfig> for PartialDapTaskConfigForReportInit<'s> {
     fn from(config: &'s DapTaskConfig) -> Self {
         PartialDapTaskConfigForReportInit {
+            not_before: config.not_before,
             not_after: config.not_after,
             method_is_taskprov: config.method_is_taskprov(),
             version: config.version,
@@ -112,6 +113,7 @@ impl<'s> From<&'s DapTaskConfig> for PartialDapTaskConfigForReportInit<'s> {
 impl<'s> From<&'s PartialDapTaskConfigForReportInit<'_>> for PartialDapTaskConfigForReportInit<'s> {
     fn from(config: &'s PartialDapTaskConfigForReportInit<'_>) -> Self {
         Self {
+            not_before: config.not_before,
             not_after: config.not_after,
             method_is_taskprov: config.method_is_taskprov,
             version: config.version,
@@ -122,6 +124,7 @@ impl<'s> From<&'s PartialDapTaskConfigForReportInit<'_>> for PartialDapTaskConfi
 }
 
 pub struct PartialDapTaskConfigForReportInit<'s> {
+    pub not_before: messages::Time,
     pub not_after: messages::Time,
     pub method_is_taskprov: bool,
     pub version: DapVersion,
@@ -152,9 +155,18 @@ impl<P> InitializedReport<P> {
             };
         }
         match report_share.report_metadata.time {
-            t if t >= task_config.not_after => reject!(TaskExpired),
+            // Reject reports that are either too far in the past or too far in the future for us
+            // to keep track of.
             t if t < valid_report_range.start => reject!(ReportDropped),
             t if valid_report_range.end < t => reject!(ReportTooEarly),
+
+            // Reject reports for tasks that haven't started or have expired.
+            //
+            // draft09 compatibility: Previously there was no notion of "start time".
+            t if task_config.version == DapVersion::Latest && t < task_config.not_before => {
+                reject!(TaskNotStarted)
+            }
+            t if t > task_config.not_after => reject!(TaskExpired),
             _ => {}
         }
 
diff --git a/crates/daphne/src/roles/leader/mod.rs b/crates/daphne/src/roles/leader/mod.rs
index c65c0a8..5882321 100644
--- a/crates/daphne/src/roles/leader/mod.rs
+++ b/crates/daphne/src/roles/leader/mod.rs
@@ -378,6 +378,7 @@ async fn run_agg_job<A: DapLeader>(
         task_config.consume_agg_job_resp(task_id, agg_job_state, agg_job_resp, metrics)?;
 
     let out_shares_count = agg_span.report_count() as u64;
+    debug!("computed {out_shares_count} output shares");
     if out_shares_count == 0 {
         return Ok(0);
     }
diff --git a/crates/daphne/src/taskprov.rs b/crates/daphne/src/taskprov.rs
index 1e41c0a..5076f84 100644
--- a/crates/daphne/src/taskprov.rs
+++ b/crates/daphne/src/taskprov.rs
@@ -12,16 +12,15 @@ use crate::{
     fatal_error,
     hpke::HpkeConfig,
     messages::{self, taskprov::TaskprovAdvertisement, Duration, TaskId, Time},
+    pine::PineParam,
     roles::aggregator::TaskprovConfig,
     vdaf::VdafVerifyKey,
-    DapBatchMode, DapError, DapTaskConfig, DapTaskConfigMethod, DapVersion, VdafConfig,
-};
-use crate::{
-    pine::PineParam,
     vdaf::{
         pine::{pine32_hmac_sha256_aes128, pine64_hmac_sha256_aes128, PineConfig},
         Prio3Config,
     },
+    DapBatchMode, DapError, DapTaskConfig, DapTaskConfigMethod, DapTaskLifetime, DapVersion,
+    VdafConfig,
 };
 use ring::{
     digest,
@@ -305,15 +304,15 @@ pub struct DapTaskConfigNeedsOptIn {
     /// Lifetime of the task.
     ///
     /// draft09: Only the expiration date is conveyed by the taskprov advertisement.
-    pub lifetime: messages::taskprov::TaskLifetime,
+    pub lifetime: DapTaskLifetime,
 }
 
 impl DapTaskConfigNeedsOptIn {
     /// Return the time after which the task is no longer valid.
     pub fn not_after(&self) -> Time {
         match self.lifetime {
-            messages::taskprov::TaskLifetime::Draft09 { expiration } => expiration,
-            messages::taskprov::TaskLifetime::Latest { start, duration } => start + duration,
+            DapTaskLifetime::Draft09 { expiration } => expiration,
+            DapTaskLifetime::Latest { start, duration } => start + duration,
         }
     }
 
@@ -367,14 +366,10 @@ impl DapTaskConfigNeedsOptIn {
     /// Complete configuration of a task via taskprov using the supplied parameters.
     pub fn into_opted_in(self, param: &OptInParam) -> DapTaskConfig {
         let (not_before, not_after) = match self.lifetime {
-            messages::taskprov::TaskLifetime::Latest { start, duration } => {
-                (start, start.saturating_add(duration))
-            }
+            DapTaskLifetime::Latest { start, duration } => (start, start.saturating_add(duration)),
             // draft09 compatibility: Previously the task start time was not conveyed by the
             // taskprov advertisement, so we need to get this value from the opt-in parameters.
-            messages::taskprov::TaskLifetime::Draft09 { expiration } => {
-                (param.not_before, expiration)
-            }
+            DapTaskLifetime::Draft09 { expiration } => (param.not_before, expiration),
         };
 
         DapTaskConfig {
@@ -477,11 +472,7 @@ impl TryFrom<&DapTaskConfig> for messages::taskprov::TaskprovAdvertisement {
                 .try_into()
                 .map_err(|_| fatal_error!(err = "task min batch size is too large for taskprov"))?,
             query_config: (&task_config.query).try_into()?,
-            lifetime: messages::taskprov::TaskLifetime::from_validity_range(
-                task_config.version,
-                task_config.not_before,
-                task_config.not_after,
-            ),
+            lifetime: task_config.lifetime(),
             vdaf_config: (&task_config.vdaf).try_into()?,
             extensions: Vec::new(),
             draft09_max_batch_query_count: match task_config.version {
@@ -510,7 +501,7 @@ mod test {
         taskprov::{DapTaskConfigNeedsOptIn, OptInParam},
         test_versions,
         vdaf::VdafVerifyKey,
-        DapRequestMeta, DapVersion,
+        DapRequestMeta, DapTaskLifetime, DapVersion,
     };
 
     /// Test conversion between the serialized task configuration and a `DapTaskConfig`.
@@ -531,7 +522,7 @@ mod test {
                     DapVersion::Latest => None,
                 },
             },
-            lifetime: messages::taskprov::TaskLifetime::from_validity_range(version, 1337, 1337),
+            lifetime: DapTaskLifetime::from_validity_range(version, 1337, 1337),
             vdaf_config: messages::taskprov::VdafConfig::Prio2 { dimension: 10 },
             extensions: Vec::new(),
             draft09_max_batch_query_count: match version {
@@ -613,7 +604,7 @@ mod test {
             query_config: messages::taskprov::QueryConfig::LeaderSelected {
                 draft09_max_batch_size: Some(NonZeroU32::new(2).unwrap()),
             },
-            lifetime: messages::taskprov::TaskLifetime::Draft09 { expiration: 23 },
+            lifetime: DapTaskLifetime::Draft09 { expiration: 23 },
             vdaf_config: messages::taskprov::VdafConfig::Prio2 { dimension: 10 },
             extensions: Vec::new(),
             draft09_max_batch_query_count: Some(23),
@@ -646,7 +637,7 @@ mod test {
             query_config: messages::taskprov::QueryConfig::LeaderSelected {
                 draft09_max_batch_size: None,
             },
-            lifetime: messages::taskprov::TaskLifetime::Latest {
+            lifetime: DapTaskLifetime::Latest {
                 start: 23,
                 duration: 23,
             },
@@ -686,7 +677,7 @@ mod test {
                         DapVersion::Latest => None,
                     },
                 },
-                lifetime: messages::taskprov::TaskLifetime::from_validity_range(version, 0, 0),
+                lifetime: DapTaskLifetime::from_validity_range(version, 0, 0),
                 // unrecognized VDAF
                 vdaf_config: messages::taskprov::VdafConfig::NotImplemented {
                     typ: 1337,
@@ -760,7 +751,7 @@ mod test {
                         DapVersion::Latest => None,
                     },
                 },
-                lifetime: messages::taskprov::TaskLifetime::from_validity_range(version, 0, 0),
+                lifetime: DapTaskLifetime::from_validity_range(version, 0, 0),
                 vdaf_config: messages::taskprov::VdafConfig::Prio2 { dimension: 1337 },
                 extensions: Vec::new(),
                 draft09_max_batch_query_count: match version {


Test patch:
diff --git a/crates/daphne-server/tests/e2e/e2e.rs b/crates/daphne-server/tests/e2e/e2e.rs
index 6163d9e..bce8d65 100644
--- a/crates/daphne-server/tests/e2e/e2e.rs
+++ b/crates/daphne-server/tests/e2e/e2e.rs
@@ -598,7 +598,7 @@ async fn internal_leader_process(version: DapVersion) {
     // Upload a number of reports (a few more than the aggregation rate).
     let mut rng = thread_rng();
     for _ in 0..t.task_config.min_batch_size + 3 {
-        let now = rng.gen_range(TestRunner::report_interval(&batch_interval));
+        let now = rng.gen_range(t.report_interval(&batch_interval));
         t.leader_request_expect_ok(
             client,
             &path,
@@ -676,7 +676,7 @@ async fn leader_collect_ok(version: DapVersion) {
     let mut time_min = u64::MAX;
     let mut time_max = 0u64;
     for _ in 0..t.task_config.min_batch_size {
-        let now = rng.gen_range(TestRunner::report_interval(&batch_interval));
+        let now = rng.gen_range(t.report_interval(&batch_interval));
         time_min = min(time_min, now);
         time_max = max(time_max, now);
         t.leader_request_expect_ok(
@@ -785,7 +785,7 @@ async fn leader_collect_ok(version: DapVersion) {
     // to avoid sharding ReportsProcessed by batch bucket, which is not feasilbe for fixed-size
     // tasks.
     //
-    //  let now = rng.gen_range(TestRunner::report_interval(&batch_interval));
+    //  let now = rng.gen_range(t.report_interval(&batch_interval));
     //  t.leader_post_expect_abort(
     //      &client,
     //      None, // dap_auth_token
@@ -819,7 +819,7 @@ async fn leader_collect_ok_interleaved(version: DapVersion) {
     // The reports are uploaded ...
     let mut rng = thread_rng();
     for _ in 0..t.task_config.min_batch_size {
-        let now = rng.gen_range(TestRunner::report_interval(&batch_interval));
+        let now = rng.gen_range(t.report_interval(&batch_interval));
         t.leader_request_expect_ok(
             client,
             &path,
@@ -883,7 +883,7 @@ async fn leader_collect_not_ready_min_batch_size(version: DapVersion) {
     // A number of reports are uploaded, but not enough to meet the minimum batch requirement.
     let mut rng = thread_rng();
     for _ in 0..t.task_config.min_batch_size - 1 {
-        let now = rng.gen_range(TestRunner::report_interval(&batch_interval));
+        let now = rng.gen_range(t.report_interval(&batch_interval));
         t.leader_request_expect_ok(
             client,
             &path,
@@ -974,7 +974,7 @@ async fn leader_collect_back_compat(version: DapVersion) {
     let mut time_min = u64::MAX;
     let mut time_max = 0u64;
     for _ in 0..t.task_config.min_batch_size {
-        let now = rng.gen_range(TestRunner::report_interval(&batch_interval));
+        let now = rng.gen_range(t.report_interval(&batch_interval));
         time_min = min(time_min, now);
         time_max = max(time_max, now);
         t.leader_request_expect_ok(
@@ -1129,7 +1129,7 @@ async fn leader_collect_abort_overlapping_batch_interval(version: DapVersion) {
     // The reports are uploaded in the background.
     let mut rng = thread_rng();
     for _ in 0..t.task_config.min_batch_size {
-        let now = rng.gen_range(TestRunner::report_interval(&batch_interval));
+        let now = rng.gen_range(t.report_interval(&batch_interval));
         t.leader_request_expect_ok(
             client,
             &path,
@@ -1433,7 +1433,7 @@ async fn leader_collect_taskprov_ok(version: DapVersion) {
     let mut rng = thread_rng();
     for _ in 0..t.task_config.min_batch_size {
         let extensions = vec![Extension::Taskprov];
-        let now = rng.gen_range(TestRunner::report_interval(&batch_interval));
+        let now = rng.gen_range(t.report_interval(&batch_interval));
         t.leader_request_expect_ok(
             client,
             &path,
diff --git a/crates/daphne-server/tests/e2e/test_runner.rs b/crates/daphne-server/tests/e2e/test_runner.rs
index 1556f85..3db0d8d 100644
--- a/crates/daphne-server/tests/e2e/test_runner.rs
+++ b/crates/daphne-server/tests/e2e/test_runner.rs
@@ -112,12 +112,15 @@ impl TestRunner {
             default_num_agg_span_shards: NonZeroUsize::new(1).unwrap(),
         };
 
+        let not_before = now;
+        let not_after = not_before + 604_800; // one week from now
+
         let task_config = DapTaskConfig {
             version,
             leader_url: leader_url.clone(),
             helper_url: helper_url.clone(),
-            not_before: now,
-            not_after: now + 604_800, // one week from now
+            not_before,
+            not_after,
             time_precision: TIME_PRECISION,
             min_batch_size: MIN_BATCH_SIZE,
             query: query_config.clone(),
@@ -240,7 +243,7 @@ impl TestRunner {
             "max_batch_size": max_batch_size,
             "time_precision": t.task_config.time_precision,
             "collector_hpke_config": collector_hpke_config_base64url.clone(),
-            "task_expiration": t.task_config.not_after,
+            "lifetime": t.task_config.lifetime(),
         });
         let add_task_path = format!("{}/internal/test/add_task", version.as_ref());
         let res: InternalTestCommandResult = t
@@ -267,7 +270,7 @@ impl TestRunner {
             "max_batch_size": max_batch_size,
             "time_precision": t.task_config.time_precision,
             "collector_hpke_config": collector_hpke_config_base64url.clone(),
-            "task_expiration": t.task_config.not_after,
+            "lifetime": t.task_config.lifetime(),
         });
         let res: InternalTestCommandResult = t
             .helper_post_internal(&add_task_path, &helper_add_task_cmd)
@@ -327,11 +330,16 @@ impl TestRunner {
         }
     }
 
-    pub fn report_interval(interval: &Interval) -> Range<u64> {
+    pub fn report_interval(&self, interval: &Interval) -> Range<u64> {
         const REPORT_STORAGE_MAX_FUTURE_TIME_SKEW: u64 = 300;
         // This is a portion of the interval which is guaranteed to be a valid report time
         // provided that the interval start time is valid.
-        interval.start..interval.start + REPORT_STORAGE_MAX_FUTURE_TIME_SKEW
+        let start = std::cmp::max(interval.start, self.task_config.not_before);
+        let end = std::cmp::min(
+            start + REPORT_STORAGE_MAX_FUTURE_TIME_SKEW,
+            self.task_config.not_after,
+        );
+        start..end
     }
 
     pub async fn get_hpke_configs(
diff --git a/crates/daphne-service-utils/src/test_route_types.rs b/crates/daphne-service-utils/src/test_route_types.rs
index 3fba74f..b611b23 100644
--- a/crates/daphne-service-utils/src/test_route_types.rs
+++ b/crates/daphne-service-utils/src/test_route_types.rs
@@ -7,8 +7,9 @@
 
 use daphne::{
     constants::DapAggregatorRole,
-    messages::{Duration, TaskId, Time},
+    messages::{Duration, TaskId},
     vdaf::{Prio3Config, VdafConfig},
+    DapTaskLifetime,
 };
 use serde::{Deserialize, Serialize};
 use std::num::NonZeroU32;
@@ -113,5 +114,7 @@ pub struct InternalTestAddTask {
     pub max_batch_size: Option<NonZeroU32>,
     pub time_precision: Duration,
     pub collector_hpke_config: String, // base64url
-    pub task_expiration: Time,
+    // TODO(cjpatton) Align this with draft-dcook-ppm-dap-interop-test-design once it's updated to
+    // DAP-13. I'm pretty sure we won't need to be backwards compatible.
+    pub lifetime: DapTaskLifetime,
 }
diff --git a/crates/daphne-worker-test/wrangler.aggregator.toml b/crates/daphne-worker-test/wrangler.aggregator.toml
index 7c15996..a969be8 100644
--- a/crates/daphne-worker-test/wrangler.aggregator.toml
+++ b/crates/daphne-worker-test/wrangler.aggregator.toml
@@ -95,6 +95,7 @@ preview_id = "24c4dc92d5cf4680e508fe18eb8f0281"
 name = "daphne-leader-aggregator"
 
 [env.leader.vars]
+DAP_TRACING="debug"
 DAP_DEPLOYMENT = "dev"
 DAP_WORKER_MODE = "aggregator"
 DAP_DURABLE_HELPER_STATE_STORE_GC_AFTER_SECS = "30"