Repository: cloudflare/daphne
Pull request #137 (2022-10-11 22:08:42)

Implement the task expiration mechanism for DAP-02
Partially addresses #100.
Based on #136 (merge that first).


Patch:
diff --git a/daphne/src/lib.rs b/daphne/src/lib.rs
index 973f527..c0c18d4 100644
--- a/daphne/src/lib.rs
+++ b/daphne/src/lib.rs
@@ -160,6 +160,11 @@ pub enum DapAbort {
     #[error("replayedReport")]
     ReplayedReport,
 
+    /// Report too late. Sent in response to an upload request for a task that is known to have
+    /// expired.
+    #[error("reportTooLate")]
+    ReportTooLate,
+
     /// Stale report. Sent in response to an upload request containing a report pertaining to a
     /// batch that has already been collected.
     #[error("staleReport")]
@@ -203,6 +208,7 @@ impl DapAbort {
             | Self::QueryMismatch
             | Self::MissingTaskId
             | Self::ReplayedReport
+            | Self::ReportTooLate
             | Self::StaleReport
             | Self::UnauthorizedRequest
             | Self::UnrecognizedAggregationJob
@@ -459,6 +465,9 @@ pub struct DapTaskConfig {
     /// constrain the batch interval of time=interval queries.
     pub time_precision: Duration,
 
+    /// The time at which the task expires.
+    pub expiration: Time,
+
     /// The query configuration for this task.
     pub query: DapQueryConfig,
 
@@ -598,6 +607,7 @@ struct ShadowDapTaskConfig {
     leader_url: Url,
     helper_url: Url,
     time_precision: Duration,
+    expiration: Time,
     query: DapQueryConfig,
     vdaf: VdafConfig,
     #[serde(with = "hex")]
@@ -618,6 +628,7 @@ impl TryFrom<ShadowDapTaskConfig> for DapTaskConfig {
             leader_url: shadow.leader_url,
             helper_url: shadow.helper_url,
             time_precision: shadow.time_precision,
+            expiration: shadow.expiration,
             query: shadow.query,
             vdaf: shadow.vdaf,
             vdaf_verify_key,
diff --git a/daphne/src/roles.rs b/daphne/src/roles.rs
index 7cdeff0..807d63a 100644
--- a/daphne/src/roles.rs
+++ b/daphne/src/roles.rs
@@ -245,6 +245,11 @@ pub trait DapLeader<'a, S>: DapAuthorizedSender<S> + DapAggregator<'a, S> {
             return Err(DapAbort::UnrecognizedHpkeConfig);
         }
 
+        // Check that the task has not expired.
+        if report.metadata.time >= task_config.as_ref().expiration {
+            return Err(DapAbort::ReportTooLate);
+        }
+
         // Store the report for future processing. At this point, the report may be rejected if
         // the Leader detects that the report was replayed or pertains to a batch that has already
         // been collected.
@@ -312,6 +317,10 @@ pub trait DapLeader<'a, S>: DapAuthorizedSender<S> + DapAggregator<'a, S> {
         // Filter out early rejected reports.
         //
         // TODO Add metrics for early rejects.
+        //
+        // TODO Add a test similar to http_post_aggregate_init_expired_task() in roles_test.rs that
+        // verifies that the Leader properly checks for expiration. This will require extending the
+        // test framework to run run_agg_job() directly.
         let early_rejects = self
             .check_early_reject(
                 task_id,
@@ -321,7 +330,12 @@ pub trait DapLeader<'a, S>: DapAuthorizedSender<S> + DapAggregator<'a, S> {
             .await?;
         let reports = reports
             .into_iter()
-            .filter(|report| early_rejects.get(&report.metadata.id).is_none())
+            .filter(|report| {
+                // Filter the reports, removing any with a timestamps beyond the expiry or that was
+                // early rejected.
+                report.metadata.time < task_config.expiration
+                    && early_rejects.get(&report.metadata.id).is_none()
+            })
             .collect();
 
         // Prepare AggregateInitializeReq.
@@ -594,14 +608,23 @@ pub trait DapHelper<'a, S>: DapAggregator<'a, S> {
                     DapHelperTransition::Continue(mut state, mut agg_resp) => {
                         let mut i = 0;
                         while i < state.seq.len() {
-                            if let Some(failure) =
-                                early_rejects.get(&agg_resp.transitions[i].report_id)
-                            {
+                            let (_vdaf_state, time, report_id) = &state.seq[i];
+
+                            let early_result =
+                                early_rejects.get(&agg_resp.transitions[i].report_id);
+                            let result = if time >= &task_config.expiration {
+                                Some(&TransitionFailure::TaskExpired)
+                            } else {
+                                early_result
+                            };
+
+                            // TODO Emit metrics for failure reasons.
+                            if let Some(failure) = result {
                                 // Mark reports that were rejected early as TransitionVar::Failed.
                                 agg_resp.transitions[i].var = TransitionVar::Failed(*failure);
 
                                 // Remove VDAF preparation state of reports that were rejected early.
-                                if state.seq[i].2 == agg_resp.transitions[i].report_id {
+                                if report_id == &agg_resp.transitions[i].report_id {
                                     let _val = state.seq.remove(i);
                                 } else {
                                     // The report ID in the Helper state and Aggregate response


Test patch:
diff --git a/daphne/src/roles_test.rs b/daphne/src/roles_test.rs
index bcb24b4..bd162d9 100644
--- a/daphne/src/roles_test.rs
+++ b/daphne/src/roles_test.rs
@@ -50,6 +50,7 @@ struct Test {
     collector_token: BearerToken,
     time_interval_task_id: Id,
     fixed_size_task_id: Id,
+    expired_task_id: Id,
 }
 
 impl Test {
@@ -82,6 +83,7 @@ impl Test {
         // Create the task list.
         let time_interval_task_id = Id(rng.gen());
         let fixed_size_task_id = Id(rng.gen());
+        let expired_task_id = Id(rng.gen());
         let mut tasks = HashMap::new();
         tasks.insert(
             time_interval_task_id.clone(),
@@ -91,6 +93,7 @@ impl Test {
                 leader_url: leader_url.clone(),
                 helper_url: helper_url.clone(),
                 time_precision,
+                expiration: now + 3600,
                 query: DapQueryConfig::TimeInterval { min_batch_size: 1 },
                 vdaf: vdaf_config.clone(),
                 vdaf_verify_key: VdafVerifyKey::Prio3(rng.gen()),
@@ -104,6 +107,7 @@ impl Test {
                 leader_url: leader_url.clone(),
                 helper_url: helper_url.clone(),
                 time_precision,
+                expiration: now + 3600,
                 query: DapQueryConfig::FixedSize {
                     min_batch_size: 1,
                     max_batch_size: 2,
@@ -112,6 +116,20 @@ impl Test {
                 vdaf_verify_key: VdafVerifyKey::Prio3(rng.gen()),
             },
         );
+        tasks.insert(
+            expired_task_id.clone(),
+            DapTaskConfig {
+                version,
+                collector_hpke_config: collector_hpke_receiver_config.config.clone(),
+                leader_url: leader_url.clone(),
+                helper_url: helper_url.clone(),
+                time_precision,
+                expiration: now, // Expires this second
+                query: DapQueryConfig::TimeInterval { min_batch_size: 1 },
+                vdaf: vdaf_config.clone(),
+                vdaf_verify_key: VdafVerifyKey::Prio3(rng.gen()),
+            },
+        );
 
         // Authorization tokens, used for all tasks.
         let leader_token = BearerToken::from("this is a bearer token!");
@@ -158,6 +176,7 @@ impl Test {
             collector_token,
             time_interval_task_id,
             fixed_size_task_id,
+            expired_task_id,
         }
     }
 
@@ -276,6 +295,9 @@ impl Test {
         report
     }
 
+    // TODO Rework the test framework to call DapLeader::run_agg_job() directly. The method here is
+    // basically a re-implementration that allows us to avoid having to mock the HTTP connection.
+    // The (major) downside is that we have to keep the code in-sync.
     async fn run_agg_job(&self, task_id: &Id) -> Result<(), DapAbort> {
         let wrapped = self.leader.get_task_config_for(task_id).await.unwrap();
         let task_config = wrapped.as_ref().unwrap();
@@ -520,6 +542,30 @@ async fn http_post_aggregate_init_unauthorized_request() {
     );
 }
 
+// Test that the Helper rejects reports past the expiration date.
+#[tokio::test]
+async fn http_post_aggregate_init_expired_task() {
+    let t = Test::new();
+
+    let report = t.gen_test_report(&t.expired_task_id).await;
+    let report_share = ReportShare {
+        metadata: report.metadata,
+        public_share: report.public_share,
+        encrypted_input_share: report.encrypted_input_shares[1].clone(),
+    };
+    let req = t
+        .gen_test_agg_init_req(&t.expired_task_id, vec![report_share])
+        .await;
+
+    let resp = t.helper.http_post_aggregate(&req).await.unwrap();
+    let agg_resp = AggregateResp::get_decoded(&resp.payload).unwrap();
+    assert_eq!(agg_resp.transitions.len(), 1);
+    assert_matches!(
+        agg_resp.transitions[0].var,
+        TransitionVar::Failed(TransitionFailure::TaskExpired)
+    );
+}
+
 #[tokio::test]
 async fn http_get_hpke_config_unrecognized_task() {
     let t = Test::new();
@@ -933,6 +979,28 @@ async fn http_post_upload_fail_send_invalid_report() {
     );
 }
 
+// Test that the Leader rejects reports past the expiration date.
+#[tokio::test]
+async fn http_post_upload_task_expired() {
+    let t = Test::new();
+    let task_id = &t.expired_task_id;
+    let task_config = t.leader.tasks.get(task_id).unwrap();
+
+    let report = t.gen_test_report(task_id).await;
+    let req = DapRequest {
+        version: task_config.version,
+        media_type: Some(MEDIA_TYPE_REPORT),
+        payload: report.get_encoded(),
+        url: task_config.leader_url.join("upload").unwrap(),
+        sender_auth: None,
+    };
+
+    assert_matches!(
+        t.leader.http_post_upload(&req).await.unwrap_err(),
+        DapAbort::ReportTooLate
+    );
+}
+
 #[tokio::test]
 async fn get_reports_empty_response() {
     let t = Test::new();
@@ -1289,9 +1357,8 @@ async fn http_post_fail_wrong_dap_version() {
     assert_matches!(err, DapAbort::InvalidProtocolVersion);
 }
 
-// Test the upload sub-protocol.
 #[tokio::test]
-async fn successful_http_post_upload() {
+async fn http_post_upload() {
     let t = Test::new();
     let task_id = &t.time_interval_task_id;
 
diff --git a/daphne_worker/src/config_test.rs b/daphne_worker/src/config_test.rs
index 271a9c2..6aca4d4 100644
--- a/daphne_worker/src/config_test.rs
+++ b/daphne_worker/src/config_test.rs
@@ -25,6 +25,7 @@ const DAP_TASK_LIST: &str = r#"{
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
         "time_precision": 3600,
+        "expiration": 1665520657,
         "query": {
             "time_interval": {
                 "min_batch_size": 100
@@ -51,6 +52,7 @@ const DAP_TASK_LIST: &str = r#"{
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
         "time_precision": 3600,
+        "expiration": 1665520657,
         "query": {
             "fixed_size": {
                 "min_batch_size": 100,
diff --git a/daphne_worker_test/helper.env b/daphne_worker_test/helper.env
index 771958d..533516a 100644
--- a/daphne_worker_test/helper.env
+++ b/daphne_worker_test/helper.env
@@ -15,7 +15,7 @@ DAP_HELPER_STATE_STORE_GARBAGE_COLLECT_AFTER_SECS = '10'
 #
 # NOTE(nakatsuka-y) The leader_url and helper_url must end with a "/".
 # When adding paths, they must not start with a "/".
-DAP_TASK_LIST = '{"f285be3caf948fcfc36b7d32181c14db95c55f04f55a2db2ee439c5879264e1f":{"version":"v02","leader_url":"http://leader:8787/v02/","helper_url":"http://helper:8788/v02/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"time_precision":3600,"query":{"time_interval":{"min_batch_size":10}},"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"1fd8d30dc0e0b7ac81f0050fcab0782d"},"6d2ef497d369fd2e72d96c47e5c28496800f5e546e912554d0df83bf24a25378":{"version":"v02","leader_url":"http://leader:8787/v02/","helper_url":"http://helper:8788/v02/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"time_precision":3600,"query":{"fixed_size":{"min_batch_size":10,"max_batch_size":12}},"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"d4c0fd634e80005dc514eb4c9538fbf7"}}'
+DAP_TASK_LIST = '{"f285be3caf948fcfc36b7d32181c14db95c55f04f55a2db2ee439c5879264e1f":{"version":"v02","leader_url":"http://leader:8787/v02/","helper_url":"http://helper:8788/v02/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"time_precision":3600,"expiration": 30695925405,"query":{"time_interval":{"min_batch_size":10}},"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"1fd8d30dc0e0b7ac81f0050fcab0782d"},"6d2ef497d369fd2e72d96c47e5c28496800f5e546e912554d0df83bf24a25378":{"version":"v02","leader_url":"http://leader:8787/v02/","helper_url":"http://helper:8788/v02/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"time_precision":3600,"expiration": 30695925405,"query":{"fixed_size":{"min_batch_size":10,"max_batch_size":12}},"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"d4c0fd634e80005dc514eb4c9538fbf7"}}'
 
 # TODO Just one token per deployment.
 DAP_LEADER_BEARER_TOKEN_LIST = '{"f285be3caf948fcfc36b7d32181c14db95c55f04f55a2db2ee439c5879264e1f":"This is an auth token ivA1e7LpnySDNn1AulaZggFLQ1n7jZ8GWOUO7GY4hgs=","6d2ef497d369fd2e72d96c47e5c28496800f5e546e912554d0df83bf24a25378":"This is yet another bearer token"}'
diff --git a/daphne_worker_test/leader.env b/daphne_worker_test/leader.env
index b6d2b71..c3c07ae 100644
--- a/daphne_worker_test/leader.env
+++ b/daphne_worker_test/leader.env
@@ -19,7 +19,7 @@ DAP_COLLECT_ID_KEY = "b416a85d280591d6da14e5b75a7d6e31"
 #
 # NOTE(nakatsuka-y) The leader_url and helper_url must end with a "/".
 # When adding paths, they must not start with a "/".
-DAP_TASK_LIST = '{"f285be3caf948fcfc36b7d32181c14db95c55f04f55a2db2ee439c5879264e1f":{"version":"v02","leader_url":"http://leader:8787/v02/","helper_url":"http://helper:8788/v02/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"time_precision":3600,"query":{"time_interval":{"min_batch_size":10}},"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"1fd8d30dc0e0b7ac81f0050fcab0782d"},"410d5e0abd94a88b8435a192cc458cc1667da2989827584cbf8a591626d5a61f":{"version":"v02","leader_url":"http://leader:8787/v02/","helper_url":"http://127.0.0.1:9788/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"time_precision":3600,"query":{"time_interval":{"min_batch_size":10}},"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"01d6232e33fe7e63b4531e3706efa8cc"},"6d2ef497d369fd2e72d96c47e5c28496800f5e546e912554d0df83bf24a25378":{"version":"v02","leader_url":"http://leader:8787/v02/","helper_url":"http://helper:8788/v02/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"time_precision":3600,"query":{"fixed_size":{"min_batch_size":10,"max_batch_size":12}},"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"d4c0fd634e80005dc514eb4c9538fbf7"}}'
+DAP_TASK_LIST = '{"f285be3caf948fcfc36b7d32181c14db95c55f04f55a2db2ee439c5879264e1f":{"version":"v02","leader_url":"http://leader:8787/v02/","helper_url":"http://helper:8788/v02/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"time_precision":3600,"expiration": 30695925405,"query":{"time_interval":{"min_batch_size":10}},"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"1fd8d30dc0e0b7ac81f0050fcab0782d"},"410d5e0abd94a88b8435a192cc458cc1667da2989827584cbf8a591626d5a61f":{"version":"v02","leader_url":"http://leader:8787/v02/","helper_url":"http://127.0.0.1:9788/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"time_precision":3600,"expiration": 30695925405,"query":{"time_interval":{"min_batch_size":10}},"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"01d6232e33fe7e63b4531e3706efa8cc"},"6d2ef497d369fd2e72d96c47e5c28496800f5e546e912554d0df83bf24a25378":{"version":"v02","leader_url":"http://leader:8787/v02/","helper_url":"http://helper:8788/v02/","collector_hpke_config":{"id":23,"kem_id":"X25519HkdfSha256","kdf_id":"HkdfSha256","aead_id":"Aes128Gcm","public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"},"time_precision":3600,"expiration": 30695925405,"query":{"fixed_size":{"min_batch_size":10,"max_batch_size":12}},"vdaf":{"prio3":{"sum":{"bits":10}}},"vdaf_verify_key":"d4c0fd634e80005dc514eb4c9538fbf7"}}'
 
 # TODO Just one token per deployment.
 DAP_LEADER_BEARER_TOKEN_LIST = '{"f285be3caf948fcfc36b7d32181c14db95c55f04f55a2db2ee439c5879264e1f":"This is an auth token ivA1e7LpnySDNn1AulaZggFLQ1n7jZ8GWOUO7GY4hgs=","410d5e0abd94a88b8435a192cc458cc1667da2989827584cbf8a591626d5a61f":"This is a differnt token 72938088f14b7ef318ef42ba72395a22","6d2ef497d369fd2e72d96c47e5c28496800f5e546e912554d0df83bf24a25378":"This is yet another bearer token"}'
diff --git a/daphne_worker_test/tests/test_runner.rs b/daphne_worker_test/tests/test_runner.rs
index 78501a1..0d2dff4 100644
--- a/daphne_worker_test/tests/test_runner.rs
+++ b/daphne_worker_test/tests/test_runner.rs
@@ -56,6 +56,7 @@ const LEADER_TASK_LIST: &str = r#"{
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
         "time_precision": 3600,
+        "expiration": 30695925405,
         "query": {
             "time_interval": {
                 "min_batch_size": 10
@@ -82,6 +83,7 @@ const LEADER_TASK_LIST: &str = r#"{
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
         "time_precision": 3600,
+        "expiration": 30695925405,
         "query": {
             "time_interval": {
                 "min_batch_size": 10
@@ -108,6 +110,7 @@ const LEADER_TASK_LIST: &str = r#"{
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
         "time_precision": 3600,
+        "expiration": 30695925405,
         "query": {
             "fixed_size": {
                 "min_batch_size": 10,
@@ -144,6 +147,7 @@ const HELPER_TASK_LIST: &str = r#"{
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
         "time_precision": 3600,
+        "expiration": 30695925405,
         "query": {
             "time_interval": {
                 "min_batch_size": 10
@@ -170,6 +174,7 @@ const HELPER_TASK_LIST: &str = r#"{
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
         "time_precision": 3600,
+        "expiration": 30695925405,
         "query": {
             "fixed_size": {
                 "min_batch_size": 10,
@@ -200,6 +205,7 @@ pub(crate) const JANUS_HELPER_TASK_LIST: &str = r#"{
             "public_key":"ec6427a49c8e9245307cc757dbdcf5d287c7a74075141af9fa566c293a52ee7c"
         },
         "time_precision": 3600,
+        "expiration": 30695925405,
         "query": {
             "time_interval": {
                 "min_batch_size": 10