Repository: cloudflare/daphne
Pull request #44 (2022-07-13 20:04:44)

Align with draft-ietf-ppm-dap-01
Based on #42 (merge that first).
Partially addresses #8.

Implements a few miscellaneous changes needed for compatibility with the next draft. In addition, this notes a few features that are known to be missing.


Patch:
diff --git a/daphne/src/lib.rs b/daphne/src/lib.rs
index 88f2168..27ade3b 100644
--- a/daphne/src/lib.rs
+++ b/daphne/src/lib.rs
@@ -13,10 +13,18 @@
 //!
 //! Daphne is not yet feature complete. Known issues include:
 //!
+//! * The collect sub-protocol has not yet been fully implemented. In particular, Daphne Aggreators
+//! do not check properly if batch intervals overlap across collect requests. Note that this
+//! feature is privacy-critical and implementation is planned. See
+//! https://github.com/cloudflare/daphne/issues/45 for details.
+//!
 //! * Daphne is not compatible with DAP tasks whose maximum batch lifetime is longer than one.
 //!
-//! * Daphne does handle aborts precisely as specified. In particular, some fields in the "Problem
-//! Details" document are missing.
+//! * Aborts are not handled precisely as specified. In particular, some fields in the "Problem
+//! Details" document are omitted.
+//!
+//! * Daphne does not implement a complete DAP Client. However, a method is provided on
+//! [`VdafConfig`](crate::VdafConfig) for producing reports.
 
 use crate::{
     messages::{CollectResp, HpkeConfig, Interval, Nonce, TransitionFailure},
@@ -127,6 +135,11 @@ pub enum DapAbort {
     #[error("unauthorizedRequest")]
     UnauthorizedRequest,
 
+    /// Unrecognized aggregation job. Sent in response to an AggregateContinueReq for which the
+    /// Helper does not recognize the indicated aggregation job.
+    #[error("unrecognizedAggregationJob")]
+    UnrecognizedAggregationJob,
+
     /// Unrecognized HPKE config. Sent in response to an upload request for which the leader share
     /// is encrypted using an unrecognized HPKE configuration.
     //
@@ -154,6 +167,7 @@ impl DapAbort {
             | Self::ReplayedReport
             | Self::StaleReport
             | Self::UnauthorizedRequest
+            | Self::UnrecognizedAggregationJob
             | Self::UnrecognizedHpkeConfig
             | Self::UnrecognizedMessage
             | Self::UnrecognizedTask => (self.to_string(), None),
diff --git a/daphne/src/messages.rs b/daphne/src/messages.rs
index 5f7dbe8..b616303 100644
--- a/daphne/src/messages.rs
+++ b/daphne/src/messages.rs
@@ -57,22 +57,22 @@ impl AsRef<[u8]> for Id {
 #[allow(missing_docs)]
 pub struct Nonce {
     pub time: u64,
-    pub rand: u64,
+    pub rand: [u8; 16],
 }
 
 impl Encode for Nonce {
     fn encode(&self, bytes: &mut Vec<u8>) {
         self.time.encode(bytes);
-        self.rand.encode(bytes);
+        bytes.extend_from_slice(&self.rand);
     }
 }
 
 impl Decode for Nonce {
     fn decode(bytes: &mut Cursor<&[u8]>) -> Result<Self, CodecError> {
-        Ok(Nonce {
-            time: u64::decode(bytes)?,
-            rand: u64::decode(bytes)?,
-        })
+        let time = u64::decode(bytes)?;
+        let mut rand = [0; 16];
+        bytes.read_exact(&mut rand)?;
+        Ok(Nonce { time, rand })
     }
 }
 
diff --git a/daphne/src/roles.rs b/daphne/src/roles.rs
index ec99adc..b8e3625 100644
--- a/daphne/src/roles.rs
+++ b/daphne/src/roles.rs
@@ -408,12 +408,13 @@ pub trait DapHelper<S>: DapAggregator<S> {
         helper_state: &DapHelperState,
     ) -> Result<(), DapError>;
 
-    /// Fetch the Helper's aggregation-flow state.
+    /// Fetch the Helper's aggregation-flow state. `None` is returned if the Helper has no state
+    /// associated with the given task and aggregation job.
     async fn get_helper_state(
         &self,
         task_id: &Id,
         agg_job_id: &Id,
-    ) -> Result<DapHelperState, DapError>;
+    ) -> Result<Option<DapHelperState>, DapError>;
 
     /// Handle an HTTP POST to `/aggregate`. The input is either an AggregateInitializeReq or
     /// AggregateContinueReq and the response is an AggregateResp.
@@ -471,7 +472,8 @@ pub trait DapHelper<S>: DapAggregator<S> {
 
                 let state = self
                     .get_helper_state(&agg_cont_req.task_id, &agg_cont_req.agg_job_id)
-                    .await?;
+                    .await?
+                    .ok_or(DapAbort::UnrecognizedAggregationJob)?;
                 let transition = task_config.vdaf.handle_agg_cont_req(state, &agg_cont_req)?;
 
                 let agg_resp = match transition {
diff --git a/daphne/src/vdaf/mod.rs b/daphne/src/vdaf/mod.rs
index fa870fd..295a428 100644
--- a/daphne/src/vdaf/mod.rs
+++ b/daphne/src/vdaf/mod.rs
@@ -104,11 +104,15 @@ impl VdafConfig {
     /// remainder belonging to the Helpers. Note that the current draft only supports one Helper,
     /// so this method will return an error if `hpke_config_list.len() != 2`.
     ///
-    /// * `now` is the number of seconds since the UNIX epoch.
+    /// * `now` is the number of seconds since the UNIX epoch. It is the caller's responsibility to
+    /// ensure this value is truncated to the nearest `min_batch_duration`, as required by the
+    /// spec.
     ///
     /// * `task_id` is the DAP task for which this report is being generated.
     ///
     /// * `measurement` is the measurement.
+    //
+    // TODO spec: Decide if truncating the timestamp should be a MAY or SHOULD.
     pub fn produce_report(
         &self,
         hpke_config_list: &[HpkeConfig],
diff --git a/daphne_worker/src/dap.rs b/daphne_worker/src/dap.rs
index 3cc4260..3cc513f 100644
--- a/daphne_worker/src/dap.rs
+++ b/daphne_worker/src/dap.rs
@@ -518,7 +518,7 @@ impl<D> DapHelper<BearerToken> for DaphneWorkerConfig<D> {
         &self,
         task_id: &Id,
         agg_job_id: &Id,
-    ) -> std::result::Result<DapHelperState, DapError> {
+    ) -> std::result::Result<Option<DapHelperState>, DapError> {
         let task_config = self
             .get_task_config_for(task_id)
             .ok_or_else(|| DapError::fatal(INT_ERR_UNRECOGNIZED_TASK))?;
@@ -528,14 +528,19 @@ impl<D> DapHelper<BearerToken> for DaphneWorkerConfig<D> {
             .id_from_name(&durable_helper_state_name(task_id, agg_job_id))?
             .get_stub()?;
 
-        let helper_state_hex: String = durable_post!(stub, DURABLE_HELPER_STATE_GET, &())
+        let res: Option<String> = durable_post!(stub, DURABLE_HELPER_STATE_GET, &())
             .await?
             .json()
             .await?;
 
-        let data = hex::decode(&helper_state_hex).map_err(|e| DapError::Fatal(e.to_string()))?;
-        let helper_state = DapHelperState::get_decoded(&task_config.vdaf, &data)?;
-
-        Ok(helper_state)
+        match res {
+            Some(helper_state_hex) => {
+                let data =
+                    hex::decode(&helper_state_hex).map_err(|e| DapError::Fatal(e.to_string()))?;
+                let helper_state = DapHelperState::get_decoded(&task_config.vdaf, &data)?;
+                Ok(Some(helper_state))
+            }
+            None => Ok(None),
+        }
     }
 }
diff --git a/daphne_worker/src/durable/helper_state_store.rs b/daphne_worker/src/durable/helper_state_store.rs
index 9c5af38..1e12b66 100644
--- a/daphne_worker/src/durable/helper_state_store.rs
+++ b/daphne_worker/src/durable/helper_state_store.rs
@@ -39,6 +39,7 @@ impl DurableObject for HelperStateStore {
                 let mut helper_state: Option<String> =
                     state_get(&self.state, "helper_state").await?;
                 if helper_state.is_some() {
+                    // TODO spec: Handle this as an abort rather than an internal error.
                     return Err(int_err("tried to overwrite helper state"));
                 }
 
@@ -52,12 +53,10 @@ impl DurableObject for HelperStateStore {
 
             (DURABLE_HELPER_STATE_GET, Method::Post) => {
                 let helper_state: Option<String> = state_get(&self.state, "helper_state").await?;
-                if let Some(helper_state) = helper_state {
+                if helper_state.is_some() {
                     self.state.storage().delete("helper_state").await?;
-                    return Response::from_json(&helper_state);
                 }
-
-                Err(int_err("tried to get helper state before it was put"))
+                Response::from_json(&helper_state)
             }
 
             _ => Err(int_err(format!(
diff --git a/daphne_worker/src/durable/report_store.rs b/daphne_worker/src/durable/report_store.rs
index 91fc6fc..ea910b5 100644
--- a/daphne_worker/src/durable/report_store.rs
+++ b/daphne_worker/src/durable/report_store.rs
@@ -145,8 +145,8 @@ pub(crate) fn nonce_hex_from_report(report_hex: &str) -> Option<&str> {
     let report_hex = &report_hex[64..];
 
     // nonce
-    if report_hex.len() < 32 {
+    if report_hex.len() < 48 {
         return None;
     }
-    Some(&report_hex[..32])
+    Some(&report_hex[..48])
 }


Test patch:
diff --git a/daphne/src/messages_test.rs b/daphne/src/messages_test.rs
index 569bcfc..cee9445 100644
--- a/daphne/src/messages_test.rs
+++ b/daphne/src/messages_test.rs
@@ -11,12 +11,12 @@ use prio::codec::{Decode, Encode};
 #[test]
 fn read_nonce() {
     let data = [
-        0, 0, 0, 0, 97, 152, 50, 20, 145, 169, 211, 83, 70, 67, 203, 171,
+        0, 0, 0, 0, 97, 152, 50, 20, 23, 23, 23, 23, 23, 23, 23, 23, 23, 23, 23, 23, 23, 23, 23, 23,
     ];
 
     let nonce = Nonce::get_decoded(&data).unwrap();
     assert_eq!(nonce.time, 1637364244);
-    assert_eq!(nonce.rand, 10496152761178246059);
+    assert_eq!(nonce.rand, [23; 16]);
 }
 
 #[test]
@@ -28,7 +28,7 @@ fn read_report() {
         ]),
         nonce: Nonce {
             time: 1637364244,
-            rand: 10496152761178246059,
+            rand: [23; 16],
         },
         extensions: vec![Extension::Unhandled {
             typ: 0xfff,
@@ -61,7 +61,7 @@ fn read_agg_init_req() {
             ReportShare {
                 nonce: Nonce {
                     time: 1637361337,
-                    rand: 10496152761178246059,
+                    rand: [99; 16],
                 },
                 extensions: Vec::default(),
                 encrypted_input_share: HpkeCiphertext {
@@ -73,7 +73,7 @@ fn read_agg_init_req() {
             ReportShare {
                 nonce: Nonce {
                     time: 163736423,
-                    rand: 123897432897439,
+                    rand: [17; 16],
                 },
                 extensions: Vec::default(),
                 encrypted_input_share: HpkeCiphertext {
@@ -98,14 +98,14 @@ fn read_agg_cont_req() {
             Transition {
                 nonce: Nonce {
                     time: 1637361337,
-                    rand: 10496152761178246059,
+                    rand: [0; 16],
                 },
                 var: TransitionVar::Continued(b"this is a VDAF-specific message".to_vec()),
             },
             Transition {
                 nonce: Nonce {
                     time: 163736423,
-                    rand: 123897432897439,
+                    rand: [1; 16],
                 },
                 var: TransitionVar::Continued(
                     b"believe it or not this is *also* a VDAF-specific message".to_vec(),
@@ -125,14 +125,14 @@ fn read_agg_resp() {
             Transition {
                 nonce: Nonce {
                     time: 1637361337,
-                    rand: 10496152761178246059,
+                    rand: [22; 16],
                 },
                 var: TransitionVar::Continued(b"this is a VDAF-specific message".to_vec()),
             },
             Transition {
                 nonce: Nonce {
                     time: 163736423,
-                    rand: 123897432897439,
+                    rand: [255; 16],
                 },
                 var: TransitionVar::Continued(
                     b"believe it or not this is *also* a VDAF-specific message".to_vec(),
diff --git a/daphne/src/roles_test.rs b/daphne/src/roles_test.rs
index d5dfa65..2224261 100644
--- a/daphne/src/roles_test.rs
+++ b/daphne/src/roles_test.rs
@@ -133,7 +133,7 @@ async fn http_post_aggregate_invalid_ciphertext() {
             report_shares: vec![ReportShare {
                 nonce: Nonce {
                     time: 1637361337,
-                    rand: 10496152761178246059,
+                    rand: [1; 16],
                 },
                 extensions: Vec::default(),
                 encrypted_input_share: HpkeCiphertext {
@@ -239,7 +239,7 @@ fn hpke_decrypter() {
         task_id: Id([23; 32]),
         nonce: Nonce {
             time: 1637364244,
-            rand: 10496152761178246059,
+            rand: [2; 16],
         },
         extensions: Vec::default(),
         encrypted_input_shares: vec![HpkeCiphertext {
diff --git a/daphne/src/testing.rs b/daphne/src/testing.rs
index 39f1a79..144fddc 100644
--- a/daphne/src/testing.rs
+++ b/daphne/src/testing.rs
@@ -219,7 +219,7 @@ impl DapHelper<BearerToken> for MockAggregator {
         &self,
         _task_id: &Id,
         _agg_job_id: &Id,
-    ) -> Result<DapHelperState, DapError> {
+    ) -> Result<Option<DapHelperState>, DapError> {
         unreachable!("not implemented");
     }
 }
diff --git a/daphne_worker/src/config_test.rs b/daphne_worker/src/config_test.rs
index 9c900a7..61ecebb 100644
--- a/daphne_worker/src/config_test.rs
+++ b/daphne_worker/src/config_test.rs
@@ -64,11 +64,11 @@ fn daphne_param() {
     // Try computing a batch name.
     let nonce = Nonce {
         time: now,
-        rand: 10496152761178246059,
+        rand: [1; 16],
     };
     assert_eq!(
         config.durable_report_store_name(&task_config, &task_id, &nonce),
-        "/task/8oW-PK-Uj8_Da30yGBwU25XFXwT1Wi2y7kOcWHkmTh8/window/1637362800/bucket/3"
+        "/task/8oW-PK-Uj8_Da30yGBwU25XFXwT1Wi2y7kOcWHkmTh8/window/1637362800/bucket/1"
     );
 
     // Try enumerating a sequence of batch names.
diff --git a/daphne_worker_test/tests/e2e.rs b/daphne_worker_test/tests/e2e.rs
index 4e892eb..9fb83de 100644
--- a/daphne_worker_test/tests/e2e.rs
+++ b/daphne_worker_test/tests/e2e.rs
@@ -8,7 +8,7 @@ mod test_runner;
 use daphne::{
     constants,
     hpke::HpkeReceiverConfig,
-    messages::{CollectReq, CollectResp, Id, Interval},
+    messages::{CollectReq, CollectResp, HpkeCiphertext, Id, Interval, Nonce, Report},
     DapAggregateResult, DapMeasurement,
 };
 use daphne_worker::InternalAggregateInfo;
@@ -131,7 +131,29 @@ async fn e2e_leader_upload() {
     let url = t.leader_url.join("/upload").unwrap();
     let resp = client
         .post(url.as_str())
-        .body(hex::decode(TEST_DATA_REPORT_OK).unwrap())
+        .body(
+            Report {
+                task_id: t.task_id.clone(),
+                nonce: Nonce {
+                    time: t.now,
+                    rand: [1; 16],
+                },
+                extensions: Vec::default(),
+                encrypted_input_shares: vec![
+                    HpkeCiphertext {
+                        config_id: 23,
+                        enc: b"encapsulated key".to_vec(),
+                        payload: b"ciphertext".to_vec(),
+                    },
+                    HpkeCiphertext {
+                        config_id: 14,
+                        enc: b"encapsulated key".to_vec(),
+                        payload: b"ciphertext".to_vec(),
+                    },
+                ],
+            }
+            .get_encoded(),
+        )
         .send()
         .await
         .expect("request failed");
@@ -446,27 +468,3 @@ async fn e2e_leader_collect_abort_invalid_batch_interval() {
     )
     .await;
 }
-
-const TEST_DATA_REPORT_OK: &str = "\
-f285be3caf948fcfc36b7d32181c14db95c55f04f55a2db2ee439c5879264e1f0000000061981e6\
-0001b75158ac9febd00000326170020d4f619a75d39dcc87d40af0a7eab96a01d19e377c0d54b5a\
-eda4b5331a1a1806024cdb4d1495ab8eeeb6c2499840758feec6de23032a199080a597f91a31355\
-2cd835a592acbd28b428b2116c7fd80cce65d9e575aacde183fcb12046acf9bb515da9dc78ad326\
-71e563b680cac5284b2e75e24af487ac504a271cd4e2ea18c6255ea281dcf6031fdb27f84a4e283\
-2b210320919a97114c14c227d970fff6f36c3d9dc181a1ff36eacddb40f710c9c0f8502ebc64baf\
-6adc57cbeb14a94f85fc4248dfcd95896cbae9ab6db070225b53e88842d1907b85bcaf4af7060cb\
-4e13bf4debdf1fcfd9760e93e6b6edad650e64cf7ae42601e03aa9893ca5b0931b7f3fc70f8abc0\
-172e4c94ed55993b85eade6ccb6f24dd9721e662e0bec03c5f7af1d51ad59a56140eba86958a1b0\
-e57dcadc82a5680e22a9e8dc4fe5f3aac134dcdfe02d9dc4b35058497242bf3edf3226b50b71e26\
-f01895e0502e953edc32efce29fef41db9d84442f6df56ea37ef8978e48c8ec0b872c9498c99742\
-f48fbfada4302768851dcda982e9eb7a816b0ca58d2e58425c53f2a90cfb9c8bdecd12f090fcdc2\
-d76b9d4406839f0c65f4ce444445025ea064a5c233c18781f7e34b74d1c2e146f5f65c11d315a65\
-efc16f294196f13e4cd51e3a25e00f94ab944479ac4c04c17eaac1e516ea36b0962073257f9566b\
-9e2643cceeaff2bb6f7166d9513004ccacf278cdade65e5b8a6e7a929b8d614703b56e0928b950a\
-f129a9d515d9dd7437cf6d5d668fbfbd7b883fbd24dc7d366aa3ae765a239de56c24d9414486719\
-2ceb7a4b3f207c54ad6fb23972b5a31d6b6d10a3664238c3733f4396a00fdfc9038f0b4bb16dbaa\
-cfef07673d00e0020e677feeec22f2fb66fd3d32143e3b6e37fe0374ad5ec511d06fee65e42b478\
-2000909f7e0d4ff40b85c324531b7773646c0669ac05ee34da63a4ee843872b3da9a3ea241ef46d\
-0049439cc78bebde3c981066d81dcb853292daf37a665d09761679a89c83477889e6ffd919ca374\
-74ed43e66ac35354a51832027ea2c165c8f86c002fe9d02220cb89ffa35251eaca6fb26287a8e09\
-03501a3e7400787de9525161af8abe2eeb1eb59b1d1446c4c98102e89";