Repository: cloudflare/daphne
Pull request #180 (2022-12-13 18:19:51)

Implement DAP-03 HpkeConfigList and new TransitionFailure values.


Patch:
diff --git a/daphne/src/constants.rs b/daphne/src/constants.rs
index fb73d92..ff4a167 100644
--- a/daphne/src/constants.rs
+++ b/daphne/src/constants.rs
@@ -7,7 +7,8 @@
 //
 // TODO spec: Decide if media type should be enforced. (We currently don't.) In any case, it may be
 // useful to enforce this for testing purposes.
-pub const MEDIA_TYPE_HPKE_CONFIG: &str = "application/dap-hpke-config";
+pub const DRAFT02_MEDIA_TYPE_HPKE_CONFIG: &str = "application/dap-hpke-config";
+pub const MEDIA_TYPE_HPKE_CONFIG_LIST: &str = "application/dap-hpke-config-list";
 pub const MEDIA_TYPE_REPORT: &str = "application/dap-report";
 pub const MEDIA_TYPE_AGG_INIT_REQ: &str = "application/dap-aggregate-initialize-req";
 pub const MEDIA_TYPE_AGG_INIT_RESP: &str = "application/dap-aggregate-initialize-resp";
@@ -22,7 +23,7 @@ pub const MEDIA_TYPE_COLLECT_RESP: &str = "application/dap-collect-resp";
 /// return a static reference to the media type.
 pub fn media_type_for(content_type: &str) -> Option<&'static str> {
     match content_type {
-        MEDIA_TYPE_HPKE_CONFIG => Some(MEDIA_TYPE_HPKE_CONFIG),
+        DRAFT02_MEDIA_TYPE_HPKE_CONFIG => Some(DRAFT02_MEDIA_TYPE_HPKE_CONFIG),
         MEDIA_TYPE_REPORT => Some(MEDIA_TYPE_REPORT),
         MEDIA_TYPE_AGG_INIT_REQ => Some(MEDIA_TYPE_AGG_INIT_REQ),
         MEDIA_TYPE_AGG_INIT_RESP => Some(MEDIA_TYPE_AGG_INIT_RESP),
diff --git a/daphne/src/messages/mod.rs b/daphne/src/messages/mod.rs
index c0d5c5d..162c587 100644
--- a/daphne/src/messages/mod.rs
+++ b/daphne/src/messages/mod.rs
@@ -512,6 +512,8 @@ pub enum TransitionFailure {
     VdafPrepError = 5,
     BatchSaturated = 6,
     TaskExpired = 7,
+    UnrecognizedMessage = 8,
+    ReportTooEarly = 9,
 }
 
 impl TryFrom<u8> for TransitionFailure {
@@ -527,6 +529,8 @@ impl TryFrom<u8> for TransitionFailure {
             b if b == Self::VdafPrepError as u8 => Ok(Self::VdafPrepError),
             b if b == Self::BatchSaturated as u8 => Ok(Self::BatchSaturated),
             b if b == Self::TaskExpired as u8 => Ok(Self::TaskExpired),
+            b if b == Self::UnrecognizedMessage as u8 => Ok(Self::UnrecognizedMessage),
+            b if b == Self::ReportTooEarly as u8 => Ok(Self::ReportTooEarly),
             _ => Err(CodecError::UnexpectedValue),
         }
     }
@@ -555,6 +559,8 @@ impl std::fmt::Display for TransitionFailure {
             Self::VdafPrepError => write!(f, "vdaf-prep-error({})", *self as u8),
             Self::BatchSaturated => write!(f, "batch-saturated({})", *self as u8),
             Self::TaskExpired => write!(f, "task-expired({})", *self as u8),
+            Self::UnrecognizedMessage => write!(f, "unrecognized-message({})", *self as u8),
+            Self::ReportTooEarly => write!(f, "report-too-early({})", *self as u8),
         }
     }
 }
@@ -959,6 +965,26 @@ impl Decode for HpkeConfig {
     }
 }
 
+/// A list of HPKE public key configurations.
+#[derive(Clone, Debug, PartialEq)]
+pub struct HpkeConfigList {
+    pub hpke_configs: Vec<HpkeConfig>,
+}
+
+impl Encode for HpkeConfigList {
+    fn encode(&self, bytes: &mut Vec<u8>) {
+        encode_u16_items(bytes, &(), &self.hpke_configs);
+    }
+}
+
+impl Decode for HpkeConfigList {
+    fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
+        Ok(Self {
+            hpke_configs: decode_u16_items(&(), bytes)?,
+        })
+    }
+}
+
 /// An HPKE ciphertext. In the DAP protocol, input shares and aggregate shares are encrypted to the
 /// intended recipient.
 #[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
diff --git a/daphne/src/roles.rs b/daphne/src/roles.rs
index cce0115..2a5caaf 100644
--- a/daphne/src/roles.rs
+++ b/daphne/src/roles.rs
@@ -5,16 +5,16 @@
 
 use crate::{
     constants::{
-        MEDIA_TYPE_AGG_CONT_REQ, MEDIA_TYPE_AGG_CONT_RESP, MEDIA_TYPE_AGG_INIT_REQ,
-        MEDIA_TYPE_AGG_INIT_RESP, MEDIA_TYPE_AGG_SHARE_REQ, MEDIA_TYPE_AGG_SHARE_RESP,
-        MEDIA_TYPE_HPKE_CONFIG,
+        DRAFT02_MEDIA_TYPE_HPKE_CONFIG, MEDIA_TYPE_AGG_CONT_REQ, MEDIA_TYPE_AGG_CONT_RESP,
+        MEDIA_TYPE_AGG_INIT_REQ, MEDIA_TYPE_AGG_INIT_RESP, MEDIA_TYPE_AGG_SHARE_REQ,
+        MEDIA_TYPE_AGG_SHARE_RESP, MEDIA_TYPE_HPKE_CONFIG_LIST,
     },
     hpke::HpkeDecrypter,
     messages::{
         constant_time_eq, AggregateContinueReq, AggregateInitializeReq, AggregateResp,
-        AggregateShareReq, AggregateShareResp, BatchSelector, CollectReq, CollectResp, Id,
-        PartialBatchSelector, Query, Report, ReportId, ReportMetadata, Time, TransitionFailure,
-        TransitionVar,
+        AggregateShareReq, AggregateShareResp, BatchSelector, CollectReq, CollectResp,
+        HpkeConfigList, Id, PartialBatchSelector, Query, Report, ReportId, ReportMetadata, Time,
+        TransitionFailure, TransitionVar,
     },
     DapAbort, DapAggregateShare, DapCollectJob, DapError, DapGlobalConfig, DapHelperState,
     DapHelperTransition, DapLeaderProcessTelemetry, DapLeaderTransition, DapOutputShare,
@@ -172,10 +172,24 @@ where
             }
         }
 
-        Ok(DapResponse {
-            media_type: Some(MEDIA_TYPE_HPKE_CONFIG),
-            payload: hpke_config.as_ref().get_encoded(),
-        })
+        match req.version {
+            DapVersion::Draft02 => Ok(DapResponse {
+                media_type: Some(DRAFT02_MEDIA_TYPE_HPKE_CONFIG),
+                payload: hpke_config.as_ref().get_encoded(),
+            }),
+            DapVersion::Draft03 => {
+                let hpke_config_list = HpkeConfigList {
+                    hpke_configs: vec![hpke_config.as_ref().clone()],
+                };
+                Ok(DapResponse {
+                    media_type: Some(MEDIA_TYPE_HPKE_CONFIG_LIST),
+                    payload: hpke_config_list.get_encoded(),
+                })
+            }
+            // This is just to keep the compiler happy as we excluded DapVersion::Unknown
+            // with an InvalidProtocolError at the top of the function.
+            DapVersion::Unknown => unreachable!("unknown DapVersion"),
+        }
     }
 
     async fn current_batch(&self, task_id: &Id) -> std::result::Result<Id, DapError>;


Test patch:
diff --git a/daphne/src/roles_test.rs b/daphne/src/roles_test.rs
index 316ef86..0aba114 100644
--- a/daphne/src/roles_test.rs
+++ b/daphne/src/roles_test.rs
@@ -5,8 +5,8 @@ use crate::{
     async_test_version, async_test_versions,
     auth::BearerToken,
     constants::{
-        MEDIA_TYPE_AGG_CONT_REQ, MEDIA_TYPE_AGG_INIT_REQ, MEDIA_TYPE_AGG_SHARE_REQ,
-        MEDIA_TYPE_COLLECT_REQ, MEDIA_TYPE_HPKE_CONFIG, MEDIA_TYPE_REPORT,
+        DRAFT02_MEDIA_TYPE_HPKE_CONFIG, MEDIA_TYPE_AGG_CONT_REQ, MEDIA_TYPE_AGG_INIT_REQ,
+        MEDIA_TYPE_AGG_SHARE_REQ, MEDIA_TYPE_COLLECT_REQ, MEDIA_TYPE_REPORT,
     },
     hpke::{HpkeDecrypter, HpkeReceiverConfig},
     messages::{
@@ -637,7 +637,7 @@ async fn http_get_hpke_config_unrecognized_task(version: DapVersion) {
     let task_id = Id(rng.gen());
     let req = DapRequest {
         version: DapVersion::Draft02,
-        media_type: Some(MEDIA_TYPE_HPKE_CONFIG),
+        media_type: Some(DRAFT02_MEDIA_TYPE_HPKE_CONFIG),
         payload: Vec::new(),
         task_id: Some(task_id.clone()),
         url: Url::parse(&format!(
@@ -660,7 +660,7 @@ async fn http_get_hpke_config_missing_task_id(version: DapVersion) {
     let t = Test::new(version);
     let req = DapRequest {
         version: DapVersion::Draft02,
-        media_type: Some(MEDIA_TYPE_HPKE_CONFIG),
+        media_type: Some(DRAFT02_MEDIA_TYPE_HPKE_CONFIG),
         task_id: Some(t.time_interval_task_id.clone()),
         payload: Vec::new(),
         url: Url::parse("http://aggregator.biz/v02/hpke_config").unwrap(),
diff --git a/daphne_worker_test/tests/e2e.rs b/daphne_worker_test/tests/e2e.rs
index 78f7a6e..73409f7 100644
--- a/daphne_worker_test/tests/e2e.rs
+++ b/daphne_worker_test/tests/e2e.rs
@@ -167,9 +167,9 @@ async fn e2e_hpke_configs_are_cached(version: DapVersion) {
     let t = TestRunner::default_with_version(version).await;
     let client = t.http_client();
     // Get a set of HPKE configs from leader and helper.
-    let hpke_config_list_0 = t.get_hpke_configs(&client).await;
+    let hpke_config_list_0 = t.get_hpke_configs(version, &client).await;
     // Get another set of HPKE configs from leader and helper.
-    let hpke_config_list_1 = t.get_hpke_configs(&client).await;
+    let hpke_config_list_1 = t.get_hpke_configs(version, &client).await;
     // The leader HPKE configs in the two sets must be the same because we store
     // the HPKE receiver config in KV.
     assert_eq!(hpke_config_list_0[0], hpke_config_list_1[0]);
@@ -183,7 +183,7 @@ async fn e2e_leader_upload(version: DapVersion) {
     let t = TestRunner::default_with_version(version).await;
     let mut rng = thread_rng();
     let client = t.http_client();
-    let hpke_config_list = t.get_hpke_configs(&client).await;
+    let hpke_config_list = t.get_hpke_configs(version, &client).await;
     let path = "upload";
 
     // Generate and upload a report.
@@ -348,7 +348,7 @@ async fn e2e_leader_upload_taskprov() {
     let version = DapVersion::Draft02;
     let t = TestRunner::default_with_version(version).await;
     let client = t.http_client();
-    let hpke_config_list = t.get_hpke_configs(&client).await;
+    let hpke_config_list = t.get_hpke_configs(version, &client).await;
     let path = "upload";
 
     // Generate and upload a report with taskprov.
@@ -517,7 +517,7 @@ async fn e2e_internal_leader_process(version: DapVersion) {
     let t = TestRunner::default_with_version(version).await;
 
     let client = t.http_client();
-    let hpke_config_list = t.get_hpke_configs(&client).await;
+    let hpke_config_list = t.get_hpke_configs(version, &client).await;
 
     let report_sel = DaphneWorkerReportSelector {
         max_agg_jobs: 100, // Needs to be sufficiently large to touch each bucket.
@@ -576,7 +576,7 @@ async fn e2e_leader_process_min_agg_rate(version: DapVersion) {
     let t = TestRunner::default_with_version(version).await;
     let client = t.http_client();
     let batch_interval = t.batch_interval();
-    let hpke_config_list = t.get_hpke_configs(&client).await;
+    let hpke_config_list = t.get_hpke_configs(version, &client).await;
 
     // The reports are uploaded in the background.
     let mut rng = thread_rng();
@@ -624,7 +624,7 @@ async fn e2e_leader_collect_ok(version: DapVersion) {
     let batch_interval = t.batch_interval();
 
     let client = t.http_client();
-    let hpke_config_list = t.get_hpke_configs(&client).await;
+    let hpke_config_list = t.get_hpke_configs(version, &client).await;
 
     // The reports are uploaded in the background.
     let mut rng = thread_rng();
@@ -749,7 +749,7 @@ async fn e2e_leader_collect_ok_interleaved(version: DapVersion) {
     let t = TestRunner::default_with_version(version).await;
     let client = t.http_client();
     let batch_interval = t.batch_interval();
-    let hpke_config_list = t.get_hpke_configs(&client).await;
+    let hpke_config_list = t.get_hpke_configs(version, &client).await;
 
     // The reports are uploaded in the background.
     let mut rng = thread_rng();
@@ -812,7 +812,7 @@ async fn e2e_leader_collect_not_ready_min_batch_size(version: DapVersion) {
     let t = TestRunner::default_with_version(version).await;
     let batch_interval = t.batch_interval();
     let client = t.http_client();
-    let hpke_config_list = t.get_hpke_configs(&client).await;
+    let hpke_config_list = t.get_hpke_configs(version, &client).await;
 
     // A number of reports are uploaded, but not enough to meet the minimum batch requirement.
     let mut rng = thread_rng();
@@ -977,7 +977,7 @@ async fn e2e_leader_collect_abort_overlapping_batch_interval(version: DapVersion
     let t = TestRunner::default_with_version(version).await;
     let batch_interval = t.batch_interval();
     let client = t.http_client();
-    let hpke_config_list = t.get_hpke_configs(&client).await;
+    let hpke_config_list = t.get_hpke_configs(version, &client).await;
 
     // The reports are uploaded in the background.
     let mut rng = thread_rng();
@@ -1081,7 +1081,7 @@ async fn e2e_fixed_size(version: DapVersion, use_current: bool) {
     };
 
     let client = t.http_client();
-    let hpke_config_list = t.get_hpke_configs(&client).await;
+    let hpke_config_list = t.get_hpke_configs(version, &client).await;
 
     // Clients: Upload reports.
     for _ in 0..t.task_config.min_batch_size {
@@ -1253,7 +1253,7 @@ async fn e2e_leader_collect_taskprov_ok(version: DapVersion) {
     let batch_interval = t.batch_interval();
 
     let client = t.http_client();
-    let hpke_config_list = t.get_hpke_configs(&client).await;
+    let hpke_config_list = t.get_hpke_configs(version, &client).await;
 
     let taskprov_task_config = TaskConfig {
         task_info: "".as_bytes().to_vec(),
diff --git a/daphne_worker_test/tests/test_runner.rs b/daphne_worker_test/tests/test_runner.rs
index 3a6738e..b6ae69a 100644
--- a/daphne_worker_test/tests/test_runner.rs
+++ b/daphne_worker_test/tests/test_runner.rs
@@ -7,7 +7,9 @@ use assert_matches::assert_matches;
 use daphne::{
     constants::MEDIA_TYPE_COLLECT_REQ,
     hpke::HpkeReceiverConfig,
-    messages::{Duration, HpkeAeadId, HpkeConfig, HpkeKdfId, HpkeKemId, Id, Interval},
+    messages::{
+        Duration, HpkeAeadId, HpkeConfig, HpkeConfigList, HpkeKdfId, HpkeKemId, Id, Interval,
+    },
     taskprov::TaskprovVersion,
     DapGlobalConfig, DapLeaderProcessTelemetry, DapQueryConfig, DapTaskConfig, DapVersion,
     Prio3Config, VdafConfig,
@@ -257,13 +259,34 @@ impl TestRunner {
         }
     }
 
-    pub async fn get_hpke_configs(&self, client: &reqwest::Client) -> [HpkeConfig; 2] {
+    pub async fn get_hpke_configs(
+        &self,
+        version: DapVersion,
+        client: &reqwest::Client,
+    ) -> [HpkeConfig; 2] {
         let raw_leader_hpke_config = self.leader_get_raw_hpke_config(&client).await;
         let raw_helper_hpke_config = self.helper_get_raw_hpke_config(&client).await;
-        [
-            HpkeConfig::get_decoded(&raw_leader_hpke_config).unwrap(),
-            HpkeConfig::get_decoded(&raw_helper_hpke_config).unwrap(),
-        ]
+        match version {
+            DapVersion::Draft02 => [
+                HpkeConfig::get_decoded(&raw_leader_hpke_config).unwrap(),
+                HpkeConfig::get_decoded(&raw_helper_hpke_config).unwrap(),
+            ],
+            _ => {
+                let mut leader_hpke_config_list =
+                    HpkeConfigList::get_decoded(&raw_leader_hpke_config).unwrap();
+                let mut helper_hpke_config_list =
+                    HpkeConfigList::get_decoded(&raw_helper_hpke_config).unwrap();
+                if leader_hpke_config_list.hpke_configs.len() != 1
+                    || helper_hpke_config_list.hpke_configs.len() != 1
+                {
+                    panic!("only a length 1 HpkeConfList is currently supported by the test suite")
+                }
+                [
+                    leader_hpke_config_list.hpke_configs.pop().unwrap(),
+                    helper_hpke_config_list.hpke_configs.pop().unwrap(),
+                ]
+            }
+        }
     }
 
     pub async fn leader_get_raw_hpke_config(&self, client: &reqwest::Client) -> Vec<u8> {