Reject reports earlier than task_config.not_before

Author: jhoyla

crates/dapf/src/main.rs

@@ -905,6 +905,10 @@ async fn handle_test_routes(action: TestAction, http_client: HttpClient) -> anyh
                         || 604_800u64,
                         "task should expire in",
                     )?,
+                task_commencement: SystemTime::now()
+                    .duration_since(SystemTime::UNIX_EPOCH)
+                    .unwrap()
+                    .as_secs(),
             };
 
             print_json(&internal_task);

crates/daphne-server/src/roles/mod.rs

@@ -111,7 +111,6 @@ mod test_utils {
         fatal_error,
         hpke::{HpkeConfig, HpkeReceiverConfig},
         messages::decode_base64url_vec,
-        roles::DapAggregator,
         vdaf::{Prio3Config, VdafConfig},
         DapBatchMode, DapError, DapTaskConfig, DapVersion,
     };
@@ -304,7 +303,7 @@ mod test_utils {
                         leader_url: cmd.leader,
                         helper_url: cmd.helper,
                         time_precision: cmd.time_precision,
-                        not_before: self.get_current_time(),
+                        not_before: cmd.task_commencement,
                         not_after: cmd.task_expiration,
                         min_batch_size: cmd.min_batch_size,
                         query,

crates/daphne-server/src/storage_proxy_connection/mod.rs

@@ -57,7 +57,7 @@ pub struct RequestBuilder<'d, B: DurableMethod, P: AsRef<[u8]>> {
     request: DurableRequest<P>,
 }
 
-impl<'d, B: DurableMethod + Debug, P: AsRef<[u8]>> RequestBuilder<'d, B, P> {
+impl<B: DurableMethod + Debug, P: AsRef<[u8]>> RequestBuilder<'_, B, P> {
     #[tracing::instrument(skip_all, fields(path = ?self.path))]
     pub async fn send<R>(self) -> Result<R, Error>
     where

crates/daphne-server/tests/e2e/e2e.rs

@@ -1490,6 +1490,12 @@ async fn leader_collect_taskprov_ok(version: DapVersion) {
     )
     .unwrap();
 
+    println!("Now - not_before: {}", t.now - task_config.not_before);
+    println!(
+        "Now - batch_interval.start: {}",
+        t.now - batch_interval.start
+    );
+    println!("t.now: {}", t.now);
     let path = TestRunner::upload_path_for_task(&task_id);
     let method = match version {
         DapVersion::Draft09 => &Method::PUT,
@@ -1500,7 +1506,9 @@ 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];
+        println!("Report interval: {:?}", TestRunner::report_interval(&batch_interval));
         let now = rng.gen_range(TestRunner::report_interval(&batch_interval));
+        println!("\tnow: {}", now);
         t.leader_request_expect_ok(
             client,
             &path,

crates/daphne-server/tests/e2e/test_runner.rs

@@ -113,7 +113,7 @@ impl TestRunner {
             version,
             leader_url: leader_url.clone(),
             helper_url: helper_url.clone(),
-            not_before: now,
+            not_before: now - (now % TIME_PRECISION) - 1,
             not_after: now + 604_800, // one week from now
             time_precision: TIME_PRECISION,
             min_batch_size: MIN_BATCH_SIZE,
@@ -236,6 +236,7 @@ impl TestRunner {
             "time_precision": t.task_config.time_precision,
             "collector_hpke_config": collector_hpke_config_base64url.clone(),
             "task_expiration": t.task_config.not_after,
+            "task_commencement": t.task_config.not_before,
         });
         let add_task_path = format!("{}/internal/test/add_task", version.as_ref());
         let res: InternalTestCommandResult = t
@@ -263,6 +264,7 @@ impl TestRunner {
             "time_precision": t.task_config.time_precision,
             "collector_hpke_config": collector_hpke_config_base64url.clone(),
             "task_expiration": t.task_config.not_after,
+            "task_commencement": t.task_config.not_before,
         });
         let res: InternalTestCommandResult = t
             .helper_post_internal(&add_task_path, &helper_add_task_cmd)

crates/daphne-service-utils/src/test_route_types.rs

@@ -113,5 +113,6 @@ pub struct InternalTestAddTask {
     pub max_batch_size: Option<NonZeroU32>,
     pub time_precision: Duration,
     pub collector_hpke_config: String, // base64url
+    pub task_commencement: Time,
     pub task_expiration: Time,
 }

crates/daphne/src/lib.rs

@@ -585,7 +585,7 @@ impl DapTaskParameters {
         )
         .unwrap()
         .into_opted_in(&taskprov::OptInParam {
-            not_before: now,
+            not_before: now - (now % self.time_precision) - 1,
             num_agg_span_shards: self.num_agg_span_shards,
         });
 

crates/daphne/src/protocol/mod.rs

@@ -274,7 +274,7 @@ mod test {
     test_versions! { produce_agg_job_req_skip_hpke_decrypt_err }
 
     fn produce_agg_job_req_skip_time_too_stale(version: DapVersion) {
-        let t = AggregationJobTest::new(TEST_VDAF, HpkeKemId::X25519HkdfSha256, version);
+        let mut t = AggregationJobTest::new(TEST_VDAF, HpkeKemId::X25519HkdfSha256, version);
         let reports = vec![t
             .task_config
             .vdaf
@@ -287,6 +287,7 @@ mod test {
             )
             .unwrap()];
 
+        t.task_config.not_before = t.valid_report_time_range().start - 2;
         let (agg_job_state, _agg_job_init_req) =
             t.produce_agg_job_req(&DapAggregationParam::Empty, reports);
         assert_eq!(agg_job_state.report_count(), 0);
@@ -298,6 +299,30 @@ mod test {
 
     test_versions! { produce_agg_job_req_skip_time_too_stale }
 
+    fn produce_agg_job_req_skip_time_before_not_before(version: DapVersion) {
+        let t = AggregationJobTest::new(TEST_VDAF, HpkeKemId::X25519HkdfSha256, version);
+        let reports = vec![t
+            .task_config
+            .vdaf
+            .produce_report(
+                &t.client_hpke_config_list,
+                t.task_config.not_before - 1,
+                &t.task_id,
+                DapMeasurement::U32Vec(vec![1; 10]),
+                t.task_config.version,
+            )
+            .unwrap()];
+        let (agg_job_state, _agg_job_init_req) =
+            t.produce_agg_job_req(&DapAggregationParam::Empty, reports);
+        assert_eq!(agg_job_state.report_count(), 0);
+
+        assert_metrics_include!(t.leader_registry, {
+            r#"report_counter{env="test_leader",host="leader.com",status="rejected_task_not_started"}"#: 1,
+        });
+    }
+
+    test_versions! {produce_agg_job_req_skip_time_before_not_before}
+
     fn produce_agg_job_req_skip_time_too_early(version: DapVersion) {
         let t = AggregationJobTest::new(TEST_VDAF, HpkeKemId::X25519HkdfSha256, version);
         let reports = vec![t
@@ -403,6 +428,7 @@ mod test {
             // 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();
+            t.task_config.not_before = t.valid_report_time_range().start - 2;
             t.valid_report_range = 0..u64::MAX;
             let (_, agg_job_init_req) =
                 t.produce_agg_job_req(&DapAggregationParam::Empty, reports.clone());
@@ -420,6 +446,40 @@ mod test {
 
     test_versions! { handle_agg_job_req_skip_time_too_stale }
 
+    fn handle_agg_job_req_skip_time_before_not_before(version: DapVersion) {
+        let mut t = AggregationJobTest::new(TEST_VDAF, HpkeKemId::X25519HkdfSha256, version);
+        let reports = vec![t
+            .task_config
+            .vdaf
+            .produce_report(
+                &t.client_hpke_config_list,
+                t.task_config.not_before - 1,
+                &t.task_id,
+                DapMeasurement::U32Vec(vec![1; 10]),
+                t.task_config.version,
+            )
+            .unwrap()];
+
+        let agg_job_init_req = {
+            // Temporarily overwrite the task_config not_before time so that the Leader accepts the
+            // out-of-range report and produces the request.
+            t.task_config.not_before -= 2;
+            let (_, agg_job_init_req) =
+                t.produce_agg_job_req(&DapAggregationParam::Empty, reports.clone());
+            t.task_config.not_before += 2;
+            agg_job_init_req
+        };
+        let (_agg_span, agg_job_resp) = t.handle_agg_job_req(agg_job_init_req);
+
+        assert_eq!(agg_job_resp.transitions.len(), 1);
+        assert_matches!(
+            agg_job_resp.transitions[0].var,
+            TransitionVar::Failed(ReportError::TaskNotStarted)
+        );
+    }
+
+    test_versions! {handle_agg_job_req_skip_time_before_not_before}
+
     fn handle_agg_job_req_skip_time_too_early(version: DapVersion) {
         let mut t = AggregationJobTest::new(TEST_VDAF, HpkeKemId::X25519HkdfSha256, version);
         let reports = vec![t

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,
@@ -145,19 +148,24 @@ impl<P> InitializedReport<P> {
         let task_config = task_config.into();
         macro_rules! reject {
             ($failure:ident) => {
+                {tracing::warn!("Rejected - {}\nTimestamp - {}", ReportError::$failure, report_share.report_metadata.time);
                 return Ok(InitializedReport::Rejected {
                     metadata: report_share.report_metadata,
                     report_err: ReportError::$failure,
-                })
+                })}
             };
         }
+
+        tracing::info!("valid_report_range: {}..{}", valid_report_range.start, valid_report_range.end);
         match report_share.report_metadata.time {
             t if t >= task_config.not_after => reject!(TaskExpired),
+            t if t < task_config.not_before => {tracing::warn!("Reject TaskNotStarted"); reject!(TaskNotStarted)},
             t if t < valid_report_range.start => reject!(ReportDropped),
             t if valid_report_range.end < t => reject!(ReportTooEarly),
             _ => {}
         }
 
+        tracing::warn!("All tests pass");
         match (
             &report_share.report_metadata.public_extensions,
             task_config.version,

crates/daphne/src/roles/leader/mod.rs

@@ -216,7 +216,6 @@ pub async fn handle_upload_req<A: DapLeader>(
         .into());
     }
 
-    // Check that the report was generated after the task's `not_before` time.
     if report.report_metadata.time
         < task_config.as_ref().not_before - task_config.as_ref().time_precision
     {
@@ -226,6 +225,13 @@ pub async fn handle_upload_req<A: DapLeader>(
         .into());
     }
 
+    // Check that the report was generated after the task's `not_before` time.
+    println!(
+        "report_metadata.time - task_config.not_before: {}",
+        report.report_metadata.time as i128 - task_config.as_ref().not_before as i128
+    );
+    println!("report_metadata.time: {}", report.report_metadata.time);
+
     if let Some(public_extensions) = &report.report_metadata.public_extensions {
         // We can be sure at this point that the ReportMetadata is well formed
         // because the decoding / error checking happens in the extractor.