Repository: cloudflare/daphne
Pull request #288 (2023-04-13 18:29:08)

Make various aborts more actionable
Partially addresses #123.


Patch:
diff --git a/daphne/src/lib.rs b/daphne/src/lib.rs
index 7e02e21..87a7c19 100644
--- a/daphne/src/lib.rs
+++ b/daphne/src/lib.rs
@@ -139,7 +139,7 @@ pub enum DapAbort {
     /// Batch overlap. Sent in response to an CollectReq for which the Leader detects the same
     /// Collector requesting an aggregate share which it has collected in the past.
     #[error("batchOverlap")]
-    BatchOverlap,
+    BatchOverlap { detail: String, task_id: TaskId },
 
     /// Internal error.
     #[error("internalError")]
@@ -161,13 +161,12 @@ pub enum DapAbort {
 
     /// Query mismatch. Sent in response to a CollectReq or AggregateShareReq.
     #[error("queryMismatch")]
-    QueryMismatch,
+    QueryMismatch { detail: String, task_id: TaskId },
 
-    /// Replayed report. Sent in response to an upload request containing a Report that has been replayed.
-    //
-    // TODO spec: Define this error type.
-    #[error("replayedReport")]
-    ReplayedReport,
+    /// Report rejected. Sent in response to an upload request containing a Report that the Leader
+    /// would reject during the aggregation sub-protocol.
+    #[error("reportReplayed")]
+    ReportRejected { detail: String },
 
     /// Report too late. Sent in response to an upload request for a task that is known to have
     /// expired.
@@ -179,11 +178,6 @@ pub enum DapAbort {
     #[error("roundMismatch")]
     RoundMismatch,
 
-    /// Stale report. Sent in response to an upload request containing a report pertaining to a
-    /// batch that has already been collected.
-    #[error("staleReport")]
-    StaleReport,
-
     /// Unauthorized HTTP request.
     #[error("unauthorizedRequest")]
     UnauthorizedRequest,
@@ -218,24 +212,23 @@ impl DapAbort {
         let (task_id, detail) = match self {
             Self::BatchInvalid { detail, task_id }
             | Self::InvalidTask { detail, task_id }
-            | Self::BatchMismatch { detail, task_id } => (Some(task_id), Some(detail)),
+            | Self::BatchMismatch { detail, task_id }
+            | Self::BatchOverlap { detail, task_id }
+            | Self::QueryMismatch { detail, task_id } => (Some(task_id), Some(detail)),
             Self::MissingTaskId => (
                 None,
                 Some("A task ID must be specified in the query parameter of the request.".into()),
             ),
-            Self::BatchOverlap
-            | Self::InvalidBatchSize
-            | Self::QueryMismatch
+            Self::ReportRejected { detail } => (None, Some(detail)),
+            Self::InvalidBatchSize
             | Self::RoundMismatch
-            | Self::ReplayedReport
             | Self::ReportTooLate
-            | Self::StaleReport
             | Self::UnauthorizedRequest
             | Self::UnrecognizedAggregationJob
             | Self::UnrecognizedHpkeConfig
             | Self::UnrecognizedMessage
             | Self::UnrecognizedTask => (None, None),
-            Self::BadRequest(s) => (None, Some(s)),
+            Self::BadRequest(detail) => (None, Some(detail)),
             Self::Internal(e) => (None, Some(e.to_string())),
         };
 
@@ -275,21 +268,60 @@ impl DapAbort {
         DapAbort::BadRequest("DAP version of request is not recognized".into())
     }
 
+    #[inline]
+    pub(crate) fn batch_overlap(task_id: &TaskId, batch_sel: &BatchSelector) -> Self {
+        Self::BatchOverlap {
+            detail: format!(
+                "The batch indicated by the request: {}",
+                serde_json::to_string(batch_sel).expect("failed to JSON-encode the batch selector while constructing a \"batchOverlap\" abort"),
+            ),
+            task_id: task_id.clone(),
+        }
+    }
+
+    #[inline]
+    pub(crate) fn query_mismatch(
+        task_id: &TaskId,
+        query_type_for_task: impl std::fmt::Display,
+        query_type_for_request: impl std::fmt::Display,
+    ) -> Self {
+        Self::QueryMismatch {
+            detail: format!("The task's query type is \"{query_type_for_task}\", but the request indicates \"{query_type_for_request}\"."),
+            task_id: task_id.clone(),
+        }
+    }
+
+    #[inline]
+    pub fn report_rejected(failure_reason: TransitionFailure) -> Self {
+        let detail = match failure_reason {
+            TransitionFailure::BatchCollected => {
+                "The report pertains to a batch that has already been collected."
+            }
+            TransitionFailure::ReportReplayed => {
+                "A report with the same ID was uploaded previously."
+            }
+            _ => return DapError::Fatal(format!("Attempted to construct a \"reportRejected\" abort with unexpected transition failure: {failure_reason:?}")).into(),
+        };
+
+        Self::ReportRejected {
+            detail: detail.into(),
+        }
+    }
+
     fn title(&self) -> Option<&'static str> {
         match self {
             Self::BatchInvalid { .. } => Some("Batch boundary check failed"),
             Self::BatchMismatch { .. } => {
                 Some("Aggregators disagree on the set of reports in the batch")
             }
-            Self::BatchOverlap => None,
+            Self::BatchOverlap { .. } => Some("The selected batch overlaps with a previous batch"),
             Self::InvalidBatchSize => None,
             Self::InvalidTask { .. } => Some("Opted out of Taskprov task"),
-            Self::QueryMismatch => None,
+            Self::QueryMismatch { .. } => Some("Query type does not match the task"),
             Self::RoundMismatch => None,
             Self::MissingTaskId => Some("Request for HPKE configuration with unspecified task"),
-            Self::ReplayedReport => None,
+            Self::ReportRejected { .. } => Some("Report rejected"),
             Self::ReportTooLate => None,
-            Self::StaleReport => None,
             Self::UnauthorizedRequest => None,
             Self::UnrecognizedAggregationJob => None,
             Self::UnrecognizedHpkeConfig => None,
@@ -305,8 +337,8 @@ impl From<DapError> for DapAbort {
     fn from(e: DapError) -> Self {
         match e {
             e @ DapError::Fatal(..) => Self::Internal(Box::new(e)),
-            DapError::Abort(e) => e,
-            DapError::Transition(t) => Self::from(t),
+            DapError::Abort(abort) => abort,
+            DapError::Transition(failure_reason) => Self::report_rejected(failure_reason),
         }
     }
 }
@@ -317,16 +349,6 @@ impl From<CodecError> for DapAbort {
     }
 }
 
-impl From<TransitionFailure> for DapAbort {
-    fn from(failure_reason: TransitionFailure) -> Self {
-        match failure_reason {
-            TransitionFailure::BatchCollected => Self::StaleReport,
-            TransitionFailure::ReportReplayed => Self::ReplayedReport,
-            _ => DapError::fatal("unhandled transition failure").into(),
-        }
-    }
-}
-
 /// A problem details document compatible with RFC 7807.
 #[derive(Debug, Deserialize, Serialize)]
 pub struct ProblemDetails {
@@ -486,6 +508,15 @@ impl DapQueryConfig {
     }
 }
 
+impl std::fmt::Display for DapQueryConfig {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::TimeInterval => write!(f, "time_interval"),
+            Self::FixedSize { .. } => write!(f, "fixed_size"),
+        }
+    }
+}
+
 /// A batch bucket.
 ///
 /// A bucket is the smallest, disjoint set of reports that can be queried: For time-interval
diff --git a/daphne/src/messages/mod.rs b/daphne/src/messages/mod.rs
index d0bcac7..c1721b7 100644
--- a/daphne/src/messages/mod.rs
+++ b/daphne/src/messages/mod.rs
@@ -292,6 +292,15 @@ pub enum PartialBatchSelector {
     FixedSizeByBatchId { batch_id: BatchId },
 }
 
+impl std::fmt::Display for PartialBatchSelector {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::TimeInterval => write!(f, "time_interval"),
+            Self::FixedSizeByBatchId { .. } => write!(f, "fixed_size"),
+        }
+    }
+}
+
 impl From<BatchSelector> for PartialBatchSelector {
     fn from(batch_sel: BatchSelector) -> Self {
         match batch_sel {
@@ -333,6 +342,15 @@ pub enum BatchSelector {
     FixedSizeByBatchId { batch_id: BatchId },
 }
 
+impl std::fmt::Display for BatchSelector {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::TimeInterval { .. } => write!(f, "time_interval"),
+            Self::FixedSizeByBatchId { .. } => write!(f, "fixed_size"),
+        }
+    }
+}
+
 impl Encode for BatchSelector {
     fn encode(&self, bytes: &mut Vec<u8>) {
         match self {
diff --git a/daphne/src/roles.rs b/daphne/src/roles.rs
index cabdeb7..adccba3 100644
--- a/daphne/src/roles.rs
+++ b/daphne/src/roles.rs
@@ -847,6 +847,7 @@ where
 
                 // Ensure we know which batch the request pertains to.
                 check_part_batch(
+                    task_id,
                     task_config,
                     &agg_job_init_req.part_batch_sel,
                     &agg_job_init_req.agg_param,
@@ -1118,12 +1119,17 @@ where
 }
 
 fn check_part_batch(
+    task_id: &TaskId,
     task_config: &DapTaskConfig,
     part_batch_sel: &PartialBatchSelector,
     agg_param: &[u8],
 ) -> Result<(), DapAbort> {
     if !task_config.query.is_valid_part_batch_sel(part_batch_sel) {
-        return Err(DapAbort::QueryMismatch);
+        return Err(DapAbort::query_mismatch(
+            task_id,
+            &task_config.query,
+            part_batch_sel,
+        ));
     }
 
     // Check that the aggreation parameter is suitable for the given VDAF.
@@ -1201,12 +1207,18 @@ where
                 });
             }
         }
-        _ => return Err(DapAbort::QueryMismatch),
+        _ => {
+            return Err(DapAbort::query_mismatch(
+                task_id,
+                &task_config.query,
+                batch_sel,
+            ))
+        }
     };
 
     // Check that the batch does not overlap with any previously collected batch.
     if batch_overlapping.await? {
-        return Err(DapAbort::BatchOverlap);
+        return Err(DapAbort::batch_overlap(task_id, batch_sel));
     }
 
     Ok(())


Test patch:
diff --git a/daphne/src/roles_test.rs b/daphne/src/roles_test.rs
index a7976f9..9663182 100644
--- a/daphne/src/roles_test.rs
+++ b/daphne/src/roles_test.rs
@@ -521,7 +521,7 @@ async fn http_post_aggregate_invalid_batch_sel(version: DapVersion) {
         .await;
     assert_matches!(
         t.helper.http_post_aggregate(&req).await.unwrap_err(),
-        DapAbort::QueryMismatch
+        DapAbort::QueryMismatch { .. }
     );
 }
 
@@ -781,7 +781,7 @@ async fn http_post_aggregate_share_invalid_batch_sel(version: DapVersion) {
         .await;
     assert_matches!(
         t.helper.http_post_aggregate_share(&req).await.unwrap_err(),
-        DapAbort::QueryMismatch
+        DapAbort::QueryMismatch { .. }
     );
 
     // Leader sends aggregate share request for unrecognized batch ID.
@@ -1395,7 +1395,7 @@ async fn http_post_collect_fail_overlapping_batch_interval(version: DapVersion)
     // run a second collect job (expect failure due to overlapping batch).
     assert_matches!(
         t.run_col_job(task_id, &query).await.unwrap_err(),
-        DapAbort::BatchOverlap
+        DapAbort::BatchOverlap { .. }
     );
 }
 
@@ -1476,7 +1476,7 @@ async fn http_post_collect_invalid_query(version: DapVersion) {
         .await;
     assert_matches!(
         t.leader.http_post_collect(&req).await.unwrap_err(),
-        DapAbort::QueryMismatch
+        DapAbort::QueryMismatch { .. }
     );
 
     // Collector indicates unrecognized batch ID.
diff --git a/daphne/src/testing.rs b/daphne/src/testing.rs
index a94c455..f31b499 100644
--- a/daphne/src/testing.rs
+++ b/daphne/src/testing.rs
@@ -490,7 +490,7 @@ where
         for bucket in task_config.batch_span_for_sel(batch_sel)? {
             if let Some(inner_agg_store) = agg_store.get(&bucket.to_owned_bucket()) {
                 if inner_agg_store.collected {
-                    return Err(DapError::Abort(DapAbort::BatchOverlap));
+                    return Err(DapError::Abort(DapAbort::batch_overlap(task_id, batch_sel)));
                 } else {
                     agg_share.merge(inner_agg_store.agg_share.clone())?;
                 }