Repository: cloudflare/daphne
Pull request #279 (2023-04-12 21:08:45)

Replace undefined "invalidProtocolVersion" abort with 400 Bad Request
Partially addresses #123.


Patch:
diff --git a/daphne/src/lib.rs b/daphne/src/lib.rs
index 711aa39..d5ae400 100644
--- a/daphne/src/lib.rs
+++ b/daphne/src/lib.rs
@@ -147,10 +147,6 @@ pub enum DapAbort {
     #[error("{0}")]
     Internal(#[source] Box<dyn std::error::Error + 'static + Send + Sync>),
 
-    /// Invalid DAP version. Sent in response to requests for an unsupported (or unknown) DAP version.
-    #[error("invalidProtocolVersion")]
-    InvalidProtocolVersion,
-
     /// Invalid batch size (either too small or too large). Sent in response to a CollectReq or
     /// AggregateShareReq.
     #[error("invalidBatchSize")]
@@ -223,7 +219,6 @@ impl DapAbort {
             | Self::BatchMismatch
             | Self::BatchOverlap
             | Self::InvalidBatchSize
-            | Self::InvalidProtocolVersion
             | Self::InvalidTask
             | Self::QueryMismatch
             | Self::RoundMismatch
@@ -262,6 +257,18 @@ impl DapAbort {
             Self::BadRequest(format!("missing content-type: expected {want_str}"))
         }
     }
+
+    #[inline]
+    pub(crate) fn version_mismatch(indicated: DapVersion, expected: DapVersion) -> Self {
+        DapAbort::BadRequest(format!(
+            "DAP version of request does not match task: got {indicated:?}; want {expected:?}"
+        ))
+    }
+
+    #[inline]
+    pub(crate) fn version_unknown() -> Self {
+        DapAbort::BadRequest("DAP version of request is not recognized".into())
+    }
 }
 
 impl From<DapError> for DapAbort {
diff --git a/daphne/src/roles.rs b/daphne/src/roles.rs
index db6e904..e5258f6 100644
--- a/daphne/src/roles.rs
+++ b/daphne/src/roles.rs
@@ -140,7 +140,7 @@ where
     ) -> Result<DapResponse, DapAbort> {
         // Check whether the DAP version indicated by the sender is supported.
         if req.version == DapVersion::Unknown {
-            return Err(DapAbort::InvalidProtocolVersion);
+            return Err(DapAbort::version_unknown());
         }
 
         // Parse the task ID from the query string, ensuring that it is the only query parameter.
@@ -167,7 +167,10 @@ where
 
             // Check whether the DAP version in the request matches the task config.
             if task_config.as_ref().version != req.version {
-                return Err(DapAbort::InvalidProtocolVersion);
+                return Err(DapAbort::version_mismatch(
+                    req.version,
+                    task_config.as_ref().version,
+                ));
             }
         }
 
@@ -179,8 +182,8 @@ where
                 };
                 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.
+            // This is just to keep the compiler happy as we excluded DapVersion::Unknown by
+            // aborting at the top of the function.
             _ => unreachable!("unhandled version {:?}", req.version),
         };
 
@@ -302,7 +305,7 @@ where
 
         // Check whether the DAP version indicated by the sender is supported.
         if req.version == DapVersion::Unknown {
-            return Err(DapAbort::InvalidProtocolVersion);
+            return Err(DapAbort::version_unknown());
         }
 
         check_request_content_type(req, DapMediaType::Report)?;
@@ -320,7 +323,10 @@ where
 
         // Check whether the DAP version in the request matches the task config.
         if task_config.as_ref().version != req.version {
-            return Err(DapAbort::InvalidProtocolVersion);
+            return Err(DapAbort::version_mismatch(
+                req.version,
+                task_config.as_ref().version,
+            ));
         }
 
         if report.encrypted_input_shares.len() != 2 {
@@ -359,7 +365,7 @@ where
 
         // Check whether the DAP version indicated by the sender is supported.
         if req.version == DapVersion::Unknown {
-            return Err(DapAbort::InvalidProtocolVersion);
+            return Err(DapAbort::version_unknown());
         }
 
         check_request_content_type(req, DapMediaType::CollectReq)?;
@@ -379,7 +385,7 @@ where
 
         // Check whether the DAP version in the request matches the task config.
         if task_config.version != req.version {
-            return Err(DapAbort::InvalidProtocolVersion);
+            return Err(DapAbort::version_mismatch(req.version, task_config.version));
         }
 
         if collect_req.query == Query::FixedSizeCurrentBatch {
@@ -754,7 +760,7 @@ where
     ) -> Result<DapResponse, DapAbort> {
         // Check whether the DAP version indicated by the sender is supported.
         if req.version == DapVersion::Unknown {
-            return Err(DapAbort::InvalidProtocolVersion);
+            return Err(DapAbort::version_unknown());
         }
 
         if !self.authorized(req).await? {
@@ -832,7 +838,7 @@ where
 
                 // Check whether the DAP version in the request matches the task config.
                 if task_config.version != req.version {
-                    return Err(DapAbort::InvalidProtocolVersion);
+                    return Err(DapAbort::version_mismatch(req.version, task_config.version));
                 }
 
                 // Ensure we know which batch the request pertains to.
@@ -941,7 +947,7 @@ where
 
                 // Check whether the DAP version in the request matches the task config.
                 if task_config.version != req.version {
-                    return Err(DapAbort::InvalidProtocolVersion);
+                    return Err(DapAbort::version_mismatch(req.version, task_config.version));
                 }
 
                 // draft02 compatibility: In draft02, the aggregation job ID is parsed from the
@@ -1017,7 +1023,7 @@ where
 
         // Check whether the DAP version indicated by the sender is supported.
         if req.version == DapVersion::Unknown {
-            return Err(DapAbort::InvalidProtocolVersion);
+            return Err(DapAbort::version_unknown());
         }
 
         check_request_content_type(req, DapMediaType::AggregateShareReq)?;
@@ -1037,7 +1043,7 @@ where
 
         // Check whether the DAP version in the request matches the task config.
         if task_config.version != req.version {
-            return Err(DapAbort::InvalidProtocolVersion);
+            return Err(DapAbort::version_mismatch(req.version, task_config.version));
         }
 
         // Ensure the batch boundaries are valid and that the batch doesn't overlap with previosuly
diff --git a/daphne_worker/src/lib.rs b/daphne_worker/src/lib.rs
index 52744d7..f6233c8 100644
--- a/daphne_worker/src/lib.rs
+++ b/daphne_worker/src/lib.rs
@@ -365,13 +365,24 @@ impl DaphneWorkerRouter {
                                 Ok(id) => id,
                                 Err(e) => return abort(e),
                             };
-                            let collect_job_id = match req.collection_job_id() {
-                                Ok(id) => id,
-                                Err(e) => return abort(e),
-                            };
+                            // We cannot check a resource here as the resource is set via
+                            // media type, and there is no media type when polling.
+                            //
+                            // We can unwrap() here as the parameter really must exist.
+                            let collect_job_id_base64url = ctx.param("collect_job_id").unwrap();
+                            let collect_job_id =
+                                match CollectionJobId::try_from_base64url(collect_job_id_base64url)
+                                {
+                                    Some(id) => id,
+                                    None => {
+                                        return abort(DapAbort::BadRequest(
+                                            "malformed collect id".into(),
+                                        ))
+                                    }
+                                };
 
                             match daph
-                                .poll_collect_job(task_id, collect_job_id)
+                                .poll_collect_job(task_id, &collect_job_id)
                                 .instrument(info_span!("poll_collect_job"))
                                 .await
                             {
diff --git a/docker/miniflare.Dockerfile b/docker/miniflare.Dockerfile
index 5f76036..f2f22f0 100644
--- a/docker/miniflare.Dockerfile
+++ b/docker/miniflare.Dockerfile
@@ -18,6 +18,7 @@ COPY daphne_worker_test ./daphne_worker_test
 COPY daphne_worker ./daphne_worker
 COPY daphne ./daphne
 WORKDIR /tmp/dap_test/daphne_worker_test
+COPY docker/wrangler.toml ./daphne_worker_test/wrangler.toml
 RUN wrangler publish --dry-run
 
 FROM alpine:3.16 AS test
@@ -25,13 +26,14 @@ RUN apk add --update npm bash
 RUN npm install -g miniflare@2.12.2
 COPY --from=builder /tmp/dap_test/daphne_worker_test/wrangler.toml /wrangler.toml
 COPY --from=builder /tmp/dap_test/daphne_worker_test/build/worker/* /build/worker/
+EXPOSE 8080
 # `-B ""` to skip build command.
 ENTRYPOINT ["miniflare", "--modules", "--modules-rule=CompiledWasm=**/*.wasm", "/build/worker/shim.mjs", "-B", ""]
 
 FROM test AS helper
 
-ENTRYPOINT ["miniflare", "--modules", "--modules-rule=CompiledWasm=**/*.wasm", "/build/worker/shim.mjs", "-B", "", "--wrangler-env=helper"]
+ENTRYPOINT ["miniflare", "--modules", "--modules-rule=CompiledWasm=**/*.wasm", "/build/worker/shim.mjs", "-B", "", "-p", "8080", "--wrangler-env=helper"]
 
 FROM test AS leader
 
-ENTRYPOINT ["miniflare", "--modules", "--modules-rule=CompiledWasm=**/*.wasm", "/build/worker/shim.mjs", "-B", "", "--wrangler-env=leader"]
+ENTRYPOINT ["miniflare", "--modules", "--modules-rule=CompiledWasm=**/*.wasm", "/build/worker/shim.mjs", "-B", "", "-p", "8080", "--wrangler-env=leader"]
diff --git a/docker/wrangler.toml b/docker/wrangler.toml
new file mode 100644
index 0000000..b7423d4
--- /dev/null
+++ b/docker/wrangler.toml
@@ -0,0 +1,170 @@
+name = "daphne_worker_test"
+main = "build/worker/shim.mjs"
+compatibility_date = "2022-10-14"
+
+[build]
+command = "cargo install --git https://github.com/cloudflare/workers-rs && worker-build"
+
+[[rules]]
+globs = ["**/*.wasm"]
+type = "CompiledWasm"
+fallthrough = false
+
+#
+# leader
+#
+
+
+[env.leader.vars]
+# NOTE: Variables marked as SECRET need to be provisioned securely in
+# production. In particular, they will not be passed as environment variables
+# as they are here. See
+# https://developers.cloudflare.com/workers/wrangler/commands/#secret.
+DAP_AGGREGATOR_ROLE = "leader"
+DAP_BASE_URL = "http://127.0.0.1:8080/"
+DAP_ISSUE73_DISABLE_AGG_JOB_QUEUE_GARBAGE_COLLECTION = "true"
+DAP_COLLECTION_JOB_ID_KEY = "b416a85d280591d6da14e5b75a7d6e31" # SECRET
+DAP_REPORT_SHARD_KEY = "61cd9685547370cfea76c2eb8d156ad9" # SECRET
+DAP_REPORT_SHARD_COUNT = "2"
+DAP_GLOBAL_CONFIG = """{
+     "report_storage_epoch_duration": 604800,
+     "report_storage_max_future_time_skew": 300,
+     "max_batch_duration": 360000,
+     "min_batch_interval_start": 259200,
+     "max_batch_interval_end": 259200,
+     "supported_hpke_kems": ["x25519_hkdf_sha256"],
+     "allow_taskprov": true,
+     "taskprov_version": "v02"
+}"""
+DAP_PROCESSED_ALARM_SAFETY_INTERVAL = "300"
+DAP_DEPLOYMENT = "dev"
+DAP_TASKPROV_HPKE_COLLECTOR_CONFIG = """{
+  "id": 23,
+  "kem_id": "p256_hkdf_sha256",
+  "kdf_id": "hkdf_sha256",
+  "aead_id": "aes128_gcm",
+  "public_key": "047dab625e0d269abcc28c611bebf5a60987ddf7e23df0e0aa343e5774ad81a1d0160d9252b82b4b5c52354205f5ec945645cb79facff8d85c9c31b490cdf35466"
+}"""
+# The secret key for the HPKE collector public key, recorded here for testing convenience:
+#     9ce9851512df3ea674b108b305c3f8c424955a94d93fd53ecf3c3f17f7d1df9e   # SECRET
+DAP_TASKPROV_VDAF_VERIFY_KEY_INIT = "b029a72fa327931a5cb643dcadcaafa098fcbfac07d990cb9e7c9a8675fafb18" # SECRET
+DAP_TASKPROV_LEADER_AUTH = """{
+  "bearer_token": "I am the leader!"
+}""" # SECRET
+DAP_TASKPROV_COLLECTOR_AUTH = """{
+  "bearer_token": "I am the collector!"
+}""" # SECRET
+DAP_DEFAULT_VERSION = "v04"
+DAP_TRACING = "debug"
+
+[env.leader.durable_objects]
+bindings = [
+    { name = "DAP_AGGREGATE_STORE", class_name = "AggregateStore" },
+    { name = "DAP_LEADER_AGG_JOB_QUEUE", class_name = "LeaderAggregationJobQueue" },
+    { name = "DAP_LEADER_BATCH_QUEUE", class_name = "LeaderBatchQueue" },
+    { name = "DAP_LEADER_COL_JOB_QUEUE", class_name = "LeaderCollectionJobQueue" },
+    { name = "DAP_GARBAGE_COLLECTOR", class_name = "GarbageCollector" },
+    { name = "DAP_REPORTS_PENDING", class_name = "ReportsPending" },
+    { name = "DAP_REPORTS_PROCESSED", class_name = "ReportsProcessed" },
+]
+
+
+[[env.leader.kv_namespaces]]
+binding = "DAP_CONFIG"
+# KV bindings are in a looked up in a namespace identified by a 16-byte id number.
+# This number is assigned by calling
+#
+#    wrangler kv:namespace create <NAME>
+#
+# for some unique name you specify, and it returns a unique id number to use.
+# Here we should use something like "leader" for the <NAME>.
+id = "<assign-one-for-the-leader>"
+# A "preview id" is an id used when running in "wrangler dev" mode locally, and
+# can just be made up.  We generated the number below with the following python
+# code:
+#
+#    import secrets
+#    print(secrets.token_hex(16))
+#
+preview_id = "24c4dc92d5cf4680e508fe18eb8f0281"
+
+
+#
+# helper
+#
+
+
+[env.helper.vars]
+# NOTE: Variables marked as SECRET need to be provisioned securely in
+# production. In particular, they will not be passed as environment variables
+# as they are here. See
+# https://developers.cloudflare.com/workers/wrangler/commands/#secret.
+DAP_ADMIN_BEARER_TOKEN = "administrator bearer token" # SECRET
+DAP_AGGREGATOR_ROLE = "helper"
+DAP_BASE_URL = "http://127.0.0.1:8080/"
+DAP_ISSUE73_DISABLE_AGG_JOB_QUEUE_GARBAGE_COLLECTION = "true"
+DAP_REPORT_SHARD_KEY = "f79c352056982bae1737e34bdac24d63" # SECRET
+DAP_REPORT_SHARD_COUNT = "2"
+DAP_HELPER_STATE_STORE_GARBAGE_COLLECT_AFTER_SECS = "10"
+DAP_GLOBAL_CONFIG = """{
+  "report_storage_epoch_duration": 604800,
+  "report_storage_max_future_time_skew": 300,
+  "max_batch_duration": 360000,
+  "min_batch_interval_start": 259200,
+  "max_batch_interval_end": 259200,
+  "supported_hpke_kems": ["x25519_hkdf_sha256"],
+  "allow_taskprov": true,
+  "taskprov_version": "v02"
+}"""
+DAP_PROCESSED_ALARM_SAFETY_INTERVAL = "300"
+DAP_DEPLOYMENT = "dev"
+DAP_TASKPROV_HPKE_COLLECTOR_CONFIG = """{
+  "id": 23,
+  "kem_id": "p256_hkdf_sha256",
+  "kdf_id": "hkdf_sha256",
+  "aead_id": "aes128_gcm",
+  "public_key": "047dab625e0d269abcc28c611bebf5a60987ddf7e23df0e0aa343e5774ad81a1d0160d9252b82b4b5c52354205f5ec945645cb79facff8d85c9c31b490cdf35466"
+}"""
+# The secret key for the HPKE collector public key, recorded here for testing convenience:
+#     9ce9851512df3ea674b108b305c3f8c424955a94d93fd53ecf3c3f17f7d1df9e   # SECRET
+DAP_TASKPROV_VDAF_VERIFY_KEY_INIT = "b029a72fa327931a5cb643dcadcaafa098fcbfac07d990cb9e7c9a8675fafb18" # SECRET
+DAP_TASKPROV_LEADER_AUTH = """{
+  "bearer_token": "I am the leader!"
+}""" # SECRET
+DAP_DEFAULT_VERSION = "v04"
+DAP_TRACING = "debug"
+
+[env.helper.durable_objects]
+bindings = [
+    { name = "DAP_AGGREGATE_STORE", class_name = "AggregateStore" },
+    { name = "DAP_HELPER_STATE_STORE", class_name = "HelperStateStore" },
+    { name = "DAP_GARBAGE_COLLECTOR", class_name = "GarbageCollector" },
+    { name = "DAP_REPORTS_PROCESSED", class_name = "ReportsProcessed" },
+]
+
+
+[[env.helper.kv_namespaces]]
+binding = "DAP_CONFIG"
+# See the comments in the leader section for the meaning of these
+# fields.  Note that a different name should be used when
+# creating the helper's namespace, as the id needs to be different from
+# the leader's or there will be errors and cross contamination if both
+# a leader and a helper exist.
+id = "<assign-one-for-the-helper>"
+# This is another unique id generated using the method for preview_id
+# described above.
+preview_id = "5d9c69a19f90be2dbcce95eb6f0a338a"
+
+
+[[migrations]]
+tag = "v1"
+new_classes = [
+    "AggregateStore",
+    "HelperStateStore",
+    "LeaderAggregationJobQueue",
+    "LeaderBatchQueue",
+    "LeaderCollectionJobQueue",
+    "GarbageCollector",
+    "ReportsPending",
+    "ReportsProcessed",
+]


Test patch:
diff --git a/daphne/src/roles_test.rs b/daphne/src/roles_test.rs
index f661a32..779c479 100644
--- a/daphne/src/roles_test.rs
+++ b/daphne/src/roles_test.rs
@@ -1508,7 +1508,7 @@ async fn http_post_collect_invalid_query(version: DapVersion) {
 async_test_versions! { http_post_collect_invalid_query }
 
 // Test HTTP POST requests with a wrong DAP version.
-async fn http_post_fail_wrong_dap_version(version: DapVersion) {
+async fn http_post_fail_unknown_version(version: DapVersion) {
     let t = Test::new(version);
     let task_id = &t.time_interval_task_id;
     let task_config = t.leader.unchecked_get_task_config(task_id).await;
@@ -1520,10 +1520,10 @@ async fn http_post_fail_wrong_dap_version(version: DapVersion) {
     req.url = task_config.leader_url.join("upload").unwrap();
 
     let err = t.leader.http_post_upload(&req).await.unwrap_err();
-    assert_matches!(err, DapAbort::InvalidProtocolVersion);
+    assert_matches!(err, DapAbort::BadRequest(details) => assert_eq!(details, "DAP version of request is not recognized"));
 }
 
-async_test_versions! { http_post_fail_wrong_dap_version }
+async_test_versions! { http_post_fail_unknown_version }
 
 async fn http_post_upload(version: DapVersion) {
     let t = Test::new(version);