Repository: cloudflare/daphne
Pull request #128 (2022-10-03 16:04:58)

Make task ID optional for HPKE config request
Partially addresses #100.
Based on #127 (merge that first).

Per DAP-02, allow the task ID to be an optional query parameter for the /hpke_config endpoint.


Patch:
diff --git a/daphne/src/hpke.rs b/daphne/src/hpke.rs
index 6d3e7df..71b4397 100644
--- a/daphne/src/hpke.rs
+++ b/daphne/src/hpke.rs
@@ -85,11 +85,11 @@ pub trait HpkeDecrypter<'a> {
     /// Return type of `get_hpke_config_for()`, wraps a reference to an HPKE config.
     type WrappedHpkeConfig: AsRef<HpkeConfig>;
 
-    /// Look up the HPKE configuration to use for the given task ID.
+    /// Look up the HPKE configuration to use for the given task ID (if specified).
     async fn get_hpke_config_for(
         &'a self,
-        task_id: &Id,
-    ) -> Result<Option<Self::WrappedHpkeConfig>, DapError>;
+        task_id: Option<&Id>,
+    ) -> Result<Self::WrappedHpkeConfig, DapError>;
 
     /// Returns `true` if a ciphertext with the HPKE config ID can be consumed in the current task.
     async fn can_hpke_decrypt(&self, task_id: &Id, config_id: u8) -> Result<bool, DapError>;
@@ -263,8 +263,8 @@ impl<'a> HpkeDecrypter<'a> for HpkeReceiverConfig {
 
     async fn get_hpke_config_for(
         &'a self,
-        _task_id: &Id,
-    ) -> Result<Option<Self::WrappedHpkeConfig>, DapError> {
+        _task_id: Option<&Id>,
+    ) -> Result<Self::WrappedHpkeConfig, DapError> {
         unreachable!("not implemented");
     }
 
diff --git a/daphne/src/lib.rs b/daphne/src/lib.rs
index 0763606..5bfae0d 100644
--- a/daphne/src/lib.rs
+++ b/daphne/src/lib.rs
@@ -138,6 +138,10 @@ pub enum DapAbort {
     #[error("invalidBatchSize")]
     InvalidBatchSize,
 
+    /// Request with missing task ID.
+    #[error("missingTaskID")]
+    MissingTaskId,
+
     /// Replayed report. Sent in response to an upload request containing a Report that has been replayed.
     //
     // TODO spec: Define this error type.
@@ -184,6 +188,7 @@ impl DapAbort {
             | Self::InvalidBatchInterval
             | Self::InvalidProtocolVersion
             | Self::InvalidBatchSize
+            | Self::MissingTaskId
             | Self::ReplayedReport
             | Self::StaleReport
             | Self::UnauthorizedRequest
diff --git a/daphne/src/roles.rs b/daphne/src/roles.rs
index 1228b6a..01f354e 100644
--- a/daphne/src/roles.rs
+++ b/daphne/src/roles.rs
@@ -163,18 +163,13 @@ pub trait DapAggregator<'a, S>: HpkeDecrypter<'a> + Sized {
             if task_config.as_ref().version != req.version {
                 return Err(DapAbort::InvalidProtocolVersion);
             }
-
-            let hpke_config = self
-                .get_hpke_config_for(task_id)
-                .await?
-                .ok_or(DapAbort::UnrecognizedTask)?;
-            Ok(DapResponse {
-                media_type: Some(MEDIA_TYPE_HPKE_CONFIG),
-                payload: hpke_config.as_ref().get_encoded(),
-            })
-        } else {
-            Err(DapAbort::BadRequest("missing query parameter".into()))
         }
+
+        let hpke_config = self.get_hpke_config_for(id.as_ref()).await?;
+        Ok(DapResponse {
+            media_type: Some(MEDIA_TYPE_HPKE_CONFIG),
+            payload: hpke_config.as_ref().get_encoded(),
+        })
     }
 }
 
diff --git a/daphne_worker/src/dap.rs b/daphne_worker/src/dap.rs
index c6e073c..0b46485 100644
--- a/daphne_worker/src/dap.rs
+++ b/daphne_worker/src/dap.rs
@@ -74,8 +74,8 @@ impl<'a, D> HpkeDecrypter<'a> for DaphneWorkerConfig<D> {
 
     async fn get_hpke_config_for(
         &'a self,
-        _task_id: &Id,
-    ) -> std::result::Result<Option<GuardedHpkeReceiverConfig<'a>>, DapError> {
+        _task_id: Option<&Id>,
+    ) -> std::result::Result<GuardedHpkeReceiverConfig<'a>, DapError> {
         let kv_store = self.kv("DAP_HPKE_RECEIVER_CONFIG_STORE").map_err(dap_err)?;
         let keys = kv_store
             .list()
@@ -120,7 +120,8 @@ impl<'a, D> HpkeDecrypter<'a> for DaphneWorkerConfig<D> {
         Ok(self
             .hpke_receiver_config(hpke_config_id)
             .await
-            .map_err(dap_err)?)
+            .map_err(dap_err)?
+            .ok_or_else(|| DapError::fatal("empty HPKE receiver config list"))?)
     }
 
     async fn can_hpke_decrypt(


Test patch:
diff --git a/daphne/src/roles_test.rs b/daphne/src/roles_test.rs
index f4dbe55..41a25f3 100644
--- a/daphne/src/roles_test.rs
+++ b/daphne/src/roles_test.rs
@@ -5,7 +5,7 @@ use crate::{
     auth::BearerToken,
     constants::{
         MEDIA_TYPE_AGG_CONT_REQ, MEDIA_TYPE_AGG_INIT_REQ, MEDIA_TYPE_AGG_SHARE_REQ,
-        MEDIA_TYPE_COLLECT_REQ, MEDIA_TYPE_REPORT,
+        MEDIA_TYPE_COLLECT_REQ, MEDIA_TYPE_HPKE_CONFIG, MEDIA_TYPE_REPORT,
     },
     hpke::HpkeReceiverConfig,
     messages::{
@@ -31,6 +31,7 @@ use std::{
     ops::DerefMut,
     vec,
 };
+use url::Url;
 
 // MockAggregator's implementation of DapLeader::get_report() always returns reports for a single
 // task. This macro is used to conveniently unwrap the task ID and reports for testing purposes.
@@ -319,6 +320,49 @@ async fn http_post_aggregate_init_unauthorized_request() {
     );
 }
 
+#[tokio::test]
+async fn http_get_hpke_config_unrecognized_task() {
+    let aggregator = MockAggregator::new();
+    let mut rng = thread_rng();
+    let task_id = Id(rng.gen());
+    let req = DapRequest {
+        version: DapVersion::Draft01,
+        media_type: Some(MEDIA_TYPE_HPKE_CONFIG),
+        payload: Vec::new(),
+        url: Url::parse(&format!(
+            "http://aggregator.biz/v01/hpke_config?task_id={}",
+            task_id.to_base64url()
+        ))
+        .unwrap(),
+        sender_auth: None,
+    };
+
+    assert_matches!(
+        aggregator.http_get_hpke_config(&req).await,
+        Err(DapAbort::UnrecognizedTask)
+    );
+}
+
+#[tokio::test]
+async fn http_get_hpke_config_missing_task_id() {
+    let aggregator = MockAggregator::new();
+    let req = DapRequest {
+        version: DapVersion::Draft01,
+        media_type: Some(MEDIA_TYPE_HPKE_CONFIG),
+        payload: Vec::new(),
+        url: Url::parse("http://aggregator.biz/v01/hpke_config").unwrap(),
+        sender_auth: None,
+    };
+
+    // An Aggregator is permitted to abort an HPKE config request if the task ID is missing. Note
+    // that Daphne-Workder does not implement this behavior. Instead it returns the HPKE config
+    // used for all tasks.
+    assert_matches!(
+        aggregator.http_get_hpke_config(&req).await,
+        Err(DapAbort::MissingTaskId)
+    );
+}
+
 #[tokio::test]
 async fn http_post_aggregate_cont_unauthorized_request() {
     let helper = MockAggregator::new();
diff --git a/daphne/src/testing.rs b/daphne/src/testing.rs
index 36d3ceb..be4e9bc 100644
--- a/daphne/src/testing.rs
+++ b/daphne/src/testing.rs
@@ -198,14 +198,23 @@ impl<'a> HpkeDecrypter<'a> for MockAggregator {
 
     async fn get_hpke_config_for(
         &'a self,
-        _task_id: &Id,
-    ) -> Result<Option<&'a HpkeConfig>, DapError> {
+        task_id: Option<&Id>,
+    ) -> Result<&'a HpkeConfig, DapError> {
         if self.hpke_receiver_config_list.is_empty() {
-            return Ok(None);
+            return Err(DapError::fatal("emtpy HPKE receiver config list"));
+        }
+
+        // Aggregators MAY abort if the HPKE config request does not specify a task ID. While not
+        // required for MockAggregator, we simulate this behavior for testing purposes.
+        //
+        // TODO(cjpatton) To make this clearer, have MockAggregator store a map from task IDs to
+        // HPKE receiver configs.
+        if task_id.is_none() {
+            return Err(DapError::Abort(DapAbort::MissingTaskId));
         }
 
         // Always advertise the first HPKE config in the list.
-        Ok(Some(&self.hpke_receiver_config_list[0].config))
+        Ok(&self.hpke_receiver_config_list[0].config)
     }
 
     async fn can_hpke_decrypt(&self, _task_id: &Id, config_id: u8) -> Result<bool, DapError> {