diff --git a/aggregator/src/aggregator.rs b/aggregator/src/aggregator.rs index 9ed4036a9..e3606e334 100644 --- a/aggregator/src/aggregator.rs +++ b/aggregator/src/aggregator.rs @@ -849,7 +849,10 @@ impl Aggregator { task_config: &TaskConfiguration, aggregator_auth_token: Option<&AuthenticationToken>, ) -> Result<(), Error> { - let (peer_aggregator, leader_url, _) = self + // The peer is always the Leader here, so `leader_url` is the peer endpoint and `helper_url` + // is this Helper's own endpoint. Both come byte-preserving from the wire TaskConfiguration. + debug_assert_eq!(peer_role, &Role::Leader); + let (peer_aggregator, leader_url, helper_url) = self .taskprov_authorize_request(peer_role, task_id, task_config, aggregator_auth_token) .await?; @@ -874,6 +877,7 @@ impl Aggregator { AggregatorTask::new( *task_id, leader_url, + helper_url, BatchMode::try_from(task_config.batch_config())?, vdaf_instance, vdaf_verify_key, @@ -894,7 +898,10 @@ impl Aggregator { )?, }, ) - .map_err(|err| Error::InvalidTask(*task_id, OptOutReason::TaskParameters(err)))?, + .map_err(|err| Error::InvalidTask(*task_id, OptOutReason::TaskParameters(err)))? + // Bind the received TaskConfiguration verbatim into HPKE AADs; it is not + // byte-reconstructible from the stored parameters. + .with_taskprov_task_config(task_config.clone()), ); self.datastore .run_tx("taskprov_put_task", |tx| { diff --git a/aggregator/src/aggregator/taskprov_tests.rs b/aggregator/src/aggregator/taskprov_tests.rs index 6b1ddf2a2..618b9f941 100644 --- a/aggregator/src/aggregator/taskprov_tests.rs +++ b/aggregator/src/aggregator/taskprov_tests.rs @@ -431,7 +431,13 @@ async fn taskprov_aggregate_init() { .eq(&AggregationJobState::AwaitingRequest) ); let got_task = got_task.unwrap(); - assert_eq!(test.task.taskprov_helper_view().unwrap(), got_task); + assert_eq!( + test.task + .taskprov_helper_view() + .unwrap() + .with_taskprov_task_config(test.task_config.clone()), + got_task + ); assert_eq!(got_task.task_info(), b"foobar".as_slice()); } @@ -596,7 +602,13 @@ async fn taskprov_aggregate_init_missing_extension() { .state() .eq(&AggregationJobState::Finished) ); - assert_eq!(test.task.taskprov_helper_view().unwrap(), got_task.unwrap()); + assert_eq!( + test.task + .taskprov_helper_view() + .unwrap() + .with_taskprov_task_config(test.task_config.clone()), + got_task.unwrap() + ); } #[tokio::test] @@ -691,7 +703,13 @@ async fn taskprov_aggregate_init_malformed_extension() { .state() .eq(&AggregationJobState::Finished) ); - assert_eq!(test.task.taskprov_helper_view().unwrap(), got_task.unwrap()); + assert_eq!( + test.task + .taskprov_helper_view() + .unwrap() + .with_taskprov_task_config(test.task_config.clone()), + got_task.unwrap() + ); } /// The helper should not opt out of the task if the current time is past the task end time. diff --git a/aggregator/src/binaries/aggregator.rs b/aggregator/src/binaries/aggregator.rs index 3216f2213..3e4a8fa2f 100644 --- a/aggregator/src/binaries/aggregator.rs +++ b/aggregator/src/binaries/aggregator.rs @@ -15,12 +15,12 @@ use educe::Educe; use janus_aggregator_api::{self, aggregator_api_handler}; use janus_aggregator_core::datastore::Datastore; use janus_core::{TokioRuntime, auth_tokens::AuthenticationToken, time::RealClock}; +use janus_messages::Url as DapUrl; use opentelemetry::metrics::Meter; use sec1::EcPrivateKey; use serde::{Deserialize, Deserializer, Serialize, de}; use tokio::{spawn, sync::watch, time::interval, try_join}; use tracing::{error, info}; -use url::Url; use crate::{ aggregator::{ @@ -250,7 +250,7 @@ pub struct AggregatorApi { /// Resource location at which the DAP service managed by this aggregator api can be found /// on the public internet. Required. #[educe(Debug(method(std::fmt::Display::fmt)))] - pub public_dap_url: Url, + pub public_dap_url: DapUrl, } fn deserialize_aggregator_api<'de, D>(deserializer: D) -> Result, D::Error> diff --git a/aggregator/src/binaries/janus_cli.rs b/aggregator/src/binaries/janus_cli.rs index 03eeb7e31..8442bd16c 100644 --- a/aggregator/src/binaries/janus_cli.rs +++ b/aggregator/src/binaries/janus_cli.rs @@ -1619,6 +1619,7 @@ mod tests { // or HPKE keys. let serialized_task_yaml = r#" - peer_aggregator_endpoint: https://helper + own_aggregator_endpoint: https://leader batch_mode: TimeInterval vdaf: !Prio3Sum max_measurement: 4096 @@ -1644,6 +1645,7 @@ mod tests { hash: MJOoBO_ysLEuG_lv2C37eEOf1Ngetsr-Ers0ZYj4vdQ hpke_keys: [] - peer_aggregator_endpoint: https://leader + own_aggregator_endpoint: https://helper batch_mode: TimeInterval aggregation_mode: Asynchronous vdaf: !Prio3Sum diff --git a/aggregator_api/src/lib.rs b/aggregator_api/src/lib.rs index 74744d7db..ed9d106ad 100644 --- a/aggregator_api/src/lib.rs +++ b/aggregator_api/src/lib.rs @@ -20,18 +20,17 @@ use janus_aggregator_core::{ http_server::{HttpMetrics, http_metrics_middleware}, }; use janus_core::{auth_tokens::AuthenticationToken, hpke, http::extract_bearer_token, time::Clock}; -use janus_messages::{HpkeConfigId, RoleParseError, TaskId}; +use janus_messages::{HpkeConfigId, RoleParseError, TaskId, Url as DapUrl}; use opentelemetry::metrics::Meter; use routes::*; use tower::ServiceBuilder; use tracing::error; -use url::Url; /// Represents the configuration for an instance of the Aggregator API. #[derive(Clone)] pub struct Config { pub auth_tokens: Vec, - pub public_dap_url: Url, + pub public_dap_url: DapUrl, } /// Content type diff --git a/aggregator_api/src/models.rs b/aggregator_api/src/models.rs index 8e9c75b7d..b5d77bf50 100644 --- a/aggregator_api/src/models.rs +++ b/aggregator_api/src/models.rs @@ -34,7 +34,7 @@ pub(crate) enum AggregatorRole { pub(crate) struct AggregatorApiConfig { pub protocol: &'static str, #[educe(Debug(method(std::fmt::Display::fmt)))] - pub dap_url: Url, + pub dap_url: DapUrl, pub role: AggregatorRole, pub vdafs: Vec, pub batch_modes: Vec, diff --git a/aggregator_api/src/routes.rs b/aggregator_api/src/routes.rs index c97e9f63d..b91325e59 100644 --- a/aggregator_api/src/routes.rs +++ b/aggregator_api/src/routes.rs @@ -205,6 +205,7 @@ pub(super) async fn post_task( AggregatorTask::new( task_id, /* peer_aggregator_endpoint */ req.peer_aggregator_endpoint, + /* own_aggregator_endpoint */ state.config.public_dap_url.clone(), /* batch_mode */ req.batch_mode, /* vdaf */ req.vdaf, vdaf_verify_key, diff --git a/aggregator_api/src/tests.rs b/aggregator_api/src/tests.rs index f86336160..2b50168b1 100644 --- a/aggregator_api/src/tests.rs +++ b/aggregator_api/src/tests.rs @@ -90,7 +90,7 @@ async fn get_config() { .unwrap(); assert!( body.contains(concat!( - r#""protocol":"DAP-18","dap_url":"https://dap.url/","role":"Either","vdafs":"#, + r#""protocol":"DAP-18","dap_url":"https://dap.url","role":"Either","vdafs":"#, r#"["Prio3Count","Prio3Sum","Prio3Histogram","Prio3SumVec"],"#, r#""batch_modes":["TimeInterval","LeaderSelected"],"#, r#""features":["TokenHash","UploadMetrics","TimeBucketedLeaderSelected","#, @@ -837,7 +837,7 @@ async fn delete_task() { let response = handler .clone() .oneshot( - Request::delete(format!("/tasks/{}", &task_id)) + Request::delete(format!("/tasks/{}", task_id)) .header("authorization", format!("Bearer {AUTH_TOKEN}")) .header("accept", CONTENT_TYPE) .body(Body::empty()) @@ -860,7 +860,7 @@ async fn delete_task() { let response = handler .clone() .oneshot( - Request::delete(format!("/tasks/{}", &task_id)) + Request::delete(format!("/tasks/{}", task_id)) .header("authorization", format!("Bearer {AUTH_TOKEN}")) .header("accept", CONTENT_TYPE) .body(Body::empty()) @@ -874,7 +874,7 @@ async fn delete_task() { let response = handler .clone() .oneshot( - Request::delete(format!("/tasks/{}", &random::())) + Request::delete(format!("/tasks/{}", random::())) .header("authorization", format!("Bearer {AUTH_TOKEN}")) .header("accept", CONTENT_TYPE) .body(Body::empty()) @@ -888,7 +888,7 @@ async fn delete_task() { let response = handler .clone() .oneshot( - Request::delete(format!("/tasks/{}", &task_id)) + Request::delete(format!("/tasks/{}", task_id)) .header("accept", CONTENT_TYPE) .body(Body::empty()) .unwrap(), @@ -1137,7 +1137,7 @@ async fn get_task_upload_metrics() { let response = handler .clone() .oneshot( - Request::get(format!("/tasks/{}/metrics/uploads", &task_id)) + Request::get(format!("/tasks/{}/metrics/uploads", task_id)) .header("authorization", format!("Bearer {AUTH_TOKEN}")) .header("accept", CONTENT_TYPE) .body(Body::empty()) @@ -1169,7 +1169,7 @@ async fn get_task_upload_metrics() { let response = handler .clone() .oneshot( - Request::get(format!("/tasks/{}/metrics/uploads", &task_id)) + Request::get(format!("/tasks/{}/metrics/uploads", task_id)) .header("authorization", format!("Bearer {AUTH_TOKEN}")) .header("accept", CONTENT_TYPE) .body(Body::empty()) @@ -1194,7 +1194,7 @@ async fn get_task_upload_metrics() { let response = handler .clone() .oneshot( - Request::get(format!("/tasks/{}/metrics/uploads", &random::())) + Request::get(format!("/tasks/{}/metrics/uploads", random::())) .header("authorization", format!("Bearer {AUTH_TOKEN}")) .header("accept", CONTENT_TYPE) .body(Body::empty()) @@ -1208,7 +1208,7 @@ async fn get_task_upload_metrics() { let response = handler .clone() .oneshot( - Request::get(format!("/tasks/{}/metrics/uploads", &task_id)) + Request::get(format!("/tasks/{}/metrics/uploads", task_id)) .header("accept", CONTENT_TYPE) .body(Body::empty()) .unwrap(), @@ -1310,7 +1310,7 @@ async fn get_task_aggregation_metrics() { .oneshot( Request::get(format!( "/tasks/{}/metrics/aggregations", - &random::() + random::() )) .header("authorization", format!("Bearer {AUTH_TOKEN}")) .header("accept", CONTENT_TYPE) @@ -2387,6 +2387,7 @@ fn task_resp_serialization() { let task = AggregatorTask::new( TaskId::from([0u8; 32]), "https://helper.com/".parse().unwrap(), + "https://leader.com/".parse().unwrap(), BatchMode::LeaderSelected { batch_time_window_size: None, }, diff --git a/aggregator_core/src/datastore.rs b/aggregator_core/src/datastore.rs index a7affbb26..c17baf68c 100644 --- a/aggregator_core/src/datastore.rs +++ b/aggregator_core/src/datastore.rs @@ -28,7 +28,8 @@ use janus_core::{ use janus_messages::{ AggregateShareId, AggregationJobId, BatchId, CollectionJobId, Duration, Extension, HpkeCiphertext, HpkeConfig, HpkeConfigId, Interval, Query, ReportId, ReportIdChecksum, - ReportMetadata, Role, TaskId, Time, TimePrecision, VerifyContinue, VerifyInit, VerifyResp, + ReportMetadata, Role, TaskConfiguration, TaskId, Time, TimePrecision, VerifyContinue, + VerifyInit, VerifyResp, batch_mode::{BatchMode, LeaderSelected, TimeInterval}, }; use leases::{acquired_aggregation_job_from_row, acquired_collection_job_from_row}; @@ -713,7 +714,8 @@ WHERE success = TRUE ORDER BY version DESC LIMIT(1)", "-- put_aggregator_task() INSERT INTO tasks ( task_id, aggregator_role, aggregation_mode, peer_aggregator_endpoint, - batch_mode, vdaf, task_start, task_duration, report_expiry_age, min_batch_size, + own_aggregator_endpoint, taskprov_task_config, batch_mode, vdaf, task_start, + task_duration, report_expiry_age, min_batch_size, time_precision, tolerable_clock_skew, collector_hpke_config, vdaf_verify_key, task_info, deactivate_at, aggregator_auth_token_type, aggregator_auth_token, aggregator_auth_token_hash, @@ -721,7 +723,7 @@ INSERT INTO tasks ( updated_at, updated_by) VALUES ( $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, - $19, $20, $21, $22, $23, $24 + $19, $20, $21, $22, $23, $24, $25, $26 ) ON CONFLICT DO NOTHING", ) @@ -735,6 +737,13 @@ ON CONFLICT DO NOTHING", /* aggregation_mode */ &task.aggregation_mode().copied(), /* peer_aggregator_endpoint */ &task.peer_aggregator_endpoint().as_str(), + /* own_aggregator_endpoint */ + &task.own_aggregator_endpoint().as_str(), + /* taskprov_task_config */ + &task + .taskprov_task_config() + .map(|c| c.get_encoded()) + .transpose()?, /* batch_mode */ &Json(task.batch_mode()), /* vdaf */ &Json(task.vdaf()), /* task_start */ @@ -889,7 +898,8 @@ UPDATE tasks SET .prepare_cached( "-- get_aggregator_task() SELECT - aggregator_role, aggregation_mode, peer_aggregator_endpoint, batch_mode, + aggregator_role, aggregation_mode, peer_aggregator_endpoint, + own_aggregator_endpoint, taskprov_task_config, batch_mode, vdaf, task_start, task_duration, report_expiry_age, min_batch_size, time_precision, tolerable_clock_skew, collector_hpke_config, vdaf_verify_key, task_info, deactivate_at, aggregator_auth_token_type, @@ -913,6 +923,7 @@ FROM tasks WHERE task_id = $1", "-- get_aggregator_tasks() SELECT task_id, aggregator_role, aggregation_mode, peer_aggregator_endpoint, + own_aggregator_endpoint, taskprov_task_config, batch_mode, vdaf, task_start, task_duration, report_expiry_age, min_batch_size, time_precision, tolerable_clock_skew, collector_hpke_config, vdaf_verify_key, task_info, deactivate_at, aggregator_auth_token_type, @@ -934,6 +945,11 @@ FROM tasks", // Scalar task parameters. let aggregator_role: AggregatorRole = row.get("aggregator_role"); let peer_aggregator_endpoint = row.get::<_, String>("peer_aggregator_endpoint").parse()?; + let own_aggregator_endpoint = row.get::<_, String>("own_aggregator_endpoint").parse()?; + let taskprov_task_config = row + .get::<_, Option>>("taskprov_task_config") + .map(|bytes| TaskConfiguration::get_decoded(&bytes)) + .transpose()?; let batch_mode = row.try_get::<_, Json>("batch_mode")?.0; let aggregation_mode: Option = row.get("aggregation_mode"); let vdaf = row.try_get::<_, Json>("vdaf")?.0; @@ -1048,6 +1064,7 @@ FROM tasks", let task = AggregatorTask::new( *task_id, peer_aggregator_endpoint, + own_aggregator_endpoint, batch_mode, vdaf, vdaf_verify_key, @@ -1060,6 +1077,10 @@ FROM tasks", aggregator_parameters, )? .with_deactivate_at(deactivate_at); + let task = match taskprov_task_config { + Some(task_config) => task.with_taskprov_task_config(task_config), + None => task, + }; Ok(task) } diff --git a/aggregator_core/src/datastore/tests.rs b/aggregator_core/src/datastore/tests.rs index 6aed85589..81b16f2b8 100644 --- a/aggregator_core/src/datastore/tests.rs +++ b/aggregator_core/src/datastore/tests.rs @@ -341,6 +341,88 @@ async fn roundtrip_task(ephemeral_datastore: EphemeralDatastore) { assert_eq!(want_tasks, got_tasks); } +#[rstest_reuse::apply(schema_versions_template)] +#[tokio::test] +async fn roundtrip_task_own_endpoint_verbatim(ephemeral_datastore: EphemeralDatastore) { + // A non-canonical own endpoint (mixed-case host, no trailing slash) must survive a DB + // round-trip byte-identical: the bytes are bound into the task's TaskConfiguration for HPKE + // AADs, and any re-encoding in the datastore write/read path would break decryption (DAP-18 + // §4.1). + install_test_trace_subscriber(); + let ds = ephemeral_datastore.datastore(MockClock::default()).await; + + let noncanonical = "https://Example.COM/DAP"; + let base = TaskBuilder::new( + task::BatchMode::TimeInterval, + AggregationMode::Synchronous, + VdafInstance::Prio3Count, + ) + .build() + .leader_view() + .unwrap(); + let task = AggregatorTask::new( + *base.id(), + base.peer_aggregator_endpoint().clone(), + noncanonical.try_into().unwrap(), + *base.batch_mode(), + base.vdaf().clone(), + base.opaque_vdaf_verify_key().clone(), + base.task_interval().copied(), + base.report_expiry_age().copied(), + base.min_batch_size(), + *base.time_precision(), + *base.tolerable_clock_skew(), + base.task_info().to_vec(), + base.aggregator_parameters().clone(), + ) + .unwrap(); + ds.put_aggregator_task(&task).await.unwrap(); + + let retrieved = ds + .run_unnamed_tx(|tx| { + let task_id = *task.id(); + Box::pin(async move { tx.get_aggregator_task(&task_id).await }) + }) + .await + .unwrap() + .unwrap(); + assert_eq!(retrieved.own_aggregator_endpoint().as_str(), noncanonical); +} + +#[rstest_reuse::apply(schema_versions_template)] +#[tokio::test] +async fn roundtrip_taskprov_task_config(ephemeral_datastore: EphemeralDatastore) { + // The verbatim taskprov TaskConfiguration must survive a DB round-trip: it is bound into HPKE + // AADs and is not byte-reconstructible from the task's other stored columns. + install_test_trace_subscriber(); + let ds = ephemeral_datastore.datastore(MockClock::default()).await; + + let base = TaskBuilder::new( + task::BatchMode::TimeInterval, + AggregationMode::Synchronous, + VdafInstance::Prio3Count, + ) + .build(); + // Source a TaskConfiguration to store from a non-taskprov view (synthesizing one for a taskprov + // task is now rejected); any valid config exercises the column's byte round-trip. + let task_config = base.leader_view().unwrap().task_configuration().unwrap(); + let task = base + .taskprov_helper_view() + .unwrap() + .with_taskprov_task_config(task_config.clone()); + ds.put_aggregator_task(&task).await.unwrap(); + + let retrieved = ds + .run_unnamed_tx(|tx| { + let task_id = *task.id(); + Box::pin(async move { tx.get_aggregator_task(&task_id).await }) + }) + .await + .unwrap() + .unwrap(); + assert_eq!(retrieved.taskprov_task_config(), Some(&task_config)); +} + #[rstest_reuse::apply(schema_versions_template)] #[tokio::test] async fn set_task_deactivate_at(ephemeral_datastore: EphemeralDatastore) { diff --git a/aggregator_core/src/task.rs b/aggregator_core/src/task.rs index e683e476f..9cc55c903 100644 --- a/aggregator_core/src/task.rs +++ b/aggregator_core/src/task.rs @@ -8,12 +8,13 @@ use chrono::{DateTime, Utc}; use educe::Educe; use janus_core::{ auth_tokens::{AuthenticationToken, AuthenticationTokenHash}, + task_config::build_task_configuration, url_for_join, vdaf::VdafInstance, }; use janus_messages::{ AggregateShareId, AggregationJobId, AggregationJobStep, BatchConfig, Duration, HpkeConfig, - Interval, Role, TaskId, Time, TimePrecision, Url as DapUrl, batch_mode, + Interval, Role, TaskConfiguration, TaskId, Time, TimePrecision, Url as DapUrl, batch_mode, }; use postgres_types::{FromSql, ToSql}; use rand::{RngExt, distr::StandardUniform, random, rng}; @@ -33,6 +34,8 @@ pub enum Error { AggregatorVerifyKeySize, #[error("base64 decode error")] Base64Decode(#[from] base64::DecodeError), + #[error(transparent)] + Message(#[from] janus_messages::Error), } /// Identifiers for batch modes used by a task, along with batch mode-specific configuration. @@ -58,7 +61,7 @@ pub enum BatchMode { impl BatchMode { /// Returns the [`BatchConfig`] message representation of this batch mode, for inclusion in a - /// [`TaskConfiguration`](janus_messages::TaskConfiguration). + /// [`TaskConfiguration`]. /// /// This is the inverse of [`BatchMode::try_from(&BatchConfig)`](TryFrom). The /// `batch_time_window_size` of [`BatchMode::LeaderSelected`] is a Janus-specific parameter that @@ -254,11 +257,22 @@ pub struct AggregatorTask { /// URL relative to which the peer aggregator's API endpoints are found. #[educe(Debug(method(std::fmt::Display::fmt)))] peer_aggregator_endpoint: DapUrl, + /// URL relative to which this aggregator's own API endpoints are found. + #[educe(Debug(method(std::fmt::Display::fmt)))] + own_aggregator_endpoint: DapUrl, /// Parameters specific to either aggregator role aggregator_parameters: AggregatorTaskParameters, /// Deactivation instant. When set and reached (per the aggregator's clock), the /// aggregator stops accepting reports for this task. This is Janus-specific state. deactivate_at: Option>, + /// For taskprov tasks, the `TaskConfiguration` exactly as received on the wire, bound verbatim + /// into HPKE AADs. `None` for API-provisioned tasks, whose `TaskConfiguration` is synthesized + /// from the stored parameters. Reconstructing a taskprov config from those parameters is not + /// byte-safe (it would drop DP strategies and unknown extensions), so the wire bytes are kept. + /// + /// Persisted only via the datastore, not [`SerializedAggregatorTask`] (like `deactivate_at`); + /// taskprov tasks are never provisioned through that serde form. + taskprov_task_config: Option, } impl AggregatorTask { @@ -267,6 +281,7 @@ impl AggregatorTask { pub fn new( task_id: TaskId, peer_aggregator_endpoint: DapUrl, + own_aggregator_endpoint: DapUrl, batch_mode: BatchMode, vdaf: VdafInstance, vdaf_verify_key: SecretBytes, @@ -293,6 +308,7 @@ impl AggregatorTask { Self::new_with_common_parameters( common_parameters, peer_aggregator_endpoint, + own_aggregator_endpoint, aggregator_parameters, ) } @@ -300,11 +316,13 @@ impl AggregatorTask { fn new_with_common_parameters( common_parameters: CommonTaskParameters, peer_aggregator_endpoint: DapUrl, + own_aggregator_endpoint: DapUrl, aggregator_parameters: AggregatorTaskParameters, ) -> Result { // Reject an unparseable endpoint at construction. Without this check a // malformed endpoint would persist and never surface a user-visible error. Url::try_from(&peer_aggregator_endpoint)?; + Url::try_from(&own_aggregator_endpoint)?; if let BatchMode::LeaderSelected { batch_time_window_size: Some(batch_time_window_size), @@ -326,8 +344,10 @@ impl AggregatorTask { Ok(Self { common_parameters, peer_aggregator_endpoint, + own_aggregator_endpoint, aggregator_parameters, deactivate_at: None, + taskprov_task_config: None, }) } @@ -351,6 +371,66 @@ impl AggregatorTask { &self.peer_aggregator_endpoint } + /// Retrieves this aggregator's own endpoint associated with this task. + pub fn own_aggregator_endpoint(&self) -> &DapUrl { + &self.own_aggregator_endpoint + } + + /// The received wire `TaskConfiguration` for a taskprov task, or `None` for an API-provisioned + /// task. + pub fn taskprov_task_config(&self) -> Option<&TaskConfiguration> { + self.taskprov_task_config.as_ref() + } + + /// Returns the canonical [`TaskConfiguration`] for this task, as bound into HPKE AADs. + /// + /// For taskprov tasks this is the wire configuration verbatim; for API-provisioned tasks it is + /// synthesized from the stored parameters, pairing this aggregator's own and peer endpoints + /// into leader/helper by role. + pub fn task_configuration(&self) -> Result { + if let Some(task_config) = &self.taskprov_task_config { + return Ok(task_config.clone()); + } + + // A taskprov task must carry its wire configuration verbatim (set in `taskprov_opt_in` and + // rehydrated on DB read); synthesizing one from the stored parameters is not byte-faithful. + // Reaching here for a taskprov task means the config was lost, so fail loudly rather than + // bind a wrong AAD. + if matches!( + self.aggregator_parameters, + AggregatorTaskParameters::TaskprovHelper { .. } + ) { + return Err(Error::InvalidParameter( + "taskprov task is missing its verbatim TaskConfiguration", + )); + } + + let (leader_endpoint, helper_endpoint) = match self.role() { + Role::Leader => ( + &self.own_aggregator_endpoint, + &self.peer_aggregator_endpoint, + ), + Role::Helper => ( + &self.peer_aggregator_endpoint, + &self.own_aggregator_endpoint, + ), + _ => return Err(Error::InvalidParameter("task role is not an aggregator")), + }; + + Ok(build_task_configuration( + self.task_info().to_vec(), + leader_endpoint.clone(), + helper_endpoint.clone(), + *self.time_precision(), + self.min_batch_size(), + self.batch_mode().to_batch_config(), + self.vdaf() + .to_vdaf_config() + .map_err(Error::InvalidParameter)?, + self.task_interval().copied(), + )?) + } + /// Retrieves the batch mode associated with this task. pub fn batch_mode(&self) -> &BatchMode { &self.common_parameters.batch_mode @@ -392,6 +472,12 @@ impl AggregatorTask { self } + /// Returns this task with its verbatim taskprov [`TaskConfiguration`] set. + pub fn with_taskprov_task_config(mut self, task_config: TaskConfiguration) -> Self { + self.taskprov_task_config = Some(task_config); + self + } + /// Retrieves the report expiry age associated with this task. pub fn report_expiry_age(&self) -> Option<&Duration> { self.common_parameters.report_expiry_age.as_ref() @@ -683,6 +769,7 @@ impl AggregatorTaskParameters { pub struct SerializedAggregatorTask { task_id: Option, peer_aggregator_endpoint: DapUrl, + own_aggregator_endpoint: DapUrl, batch_mode: BatchMode, aggregation_mode: Option, vdaf: VdafInstance, @@ -743,6 +830,7 @@ impl Serialize for AggregatorTask { SerializedAggregatorTask { task_id: Some(*self.id()), peer_aggregator_endpoint: self.peer_aggregator_endpoint().clone(), + own_aggregator_endpoint: self.own_aggregator_endpoint().clone(), batch_mode: *self.batch_mode(), aggregation_mode: self.aggregator_parameters.aggregation_mode().copied(), vdaf: self.vdaf().clone(), @@ -837,6 +925,7 @@ impl TryFrom for AggregatorTask { AggregatorTask::new( task_id, serialized_task.peer_aggregator_endpoint, + serialized_task.own_aggregator_endpoint, serialized_task.batch_mode, serialized_task.vdaf, SecretBytes::new(URL_SAFE_NO_PAD.decode(vdaf_verify_key)?), @@ -1138,6 +1227,7 @@ pub mod test_util { AggregatorTask::new_with_common_parameters( self.common_parameters.clone(), to_dap_url(&self.helper_aggregator_endpoint), + to_dap_url(&self.leader_aggregator_endpoint), AggregatorTaskParameters::Leader { aggregator_auth_token: self.aggregator_auth_token.clone(), collector_auth_token_hash: AuthenticationTokenHash::from( @@ -1153,6 +1243,7 @@ pub mod test_util { AggregatorTask::new_with_common_parameters( self.common_parameters.clone(), to_dap_url(&self.leader_aggregator_endpoint), + to_dap_url(&self.helper_aggregator_endpoint), AggregatorTaskParameters::Helper { aggregator_auth_token_hash: AuthenticationTokenHash::from( &self.aggregator_auth_token, @@ -1168,6 +1259,7 @@ pub mod test_util { AggregatorTask::new_with_common_parameters( self.common_parameters.clone(), to_dap_url(&self.leader_aggregator_endpoint), + to_dap_url(&self.helper_aggregator_endpoint), AggregatorTaskParameters::TaskprovHelper { aggregation_mode: self.helper_aggregation_mode, }, @@ -1423,6 +1515,16 @@ pub mod test_util { self } + /// Gets the `task_info` field for the eventual task. + pub fn task_info(&self) -> &[u8] { + &self.0.common_parameters.task_info + } + + /// Gets the task interval for the eventual task. + pub fn task_interval(&self) -> Option<&Interval> { + self.0.common_parameters.task_interval.as_ref() + } + /// Gets the colector HPKE keypair for the eventual task. pub fn collector_hpke_keypair(&self) -> &HpkeKeypair { self.0.collector_hpke_keypair() @@ -1451,12 +1553,13 @@ mod tests { use chrono::TimeDelta; use janus_core::{ auth_tokens::{AuthenticationToken, AuthenticationTokenHash}, + task_config::build_task_configuration, test_util::roundtrip_encoding, vdaf::vdaf_dp_strategies, }; use janus_messages::{ BatchConfig, Duration, HpkeAeadId, HpkeConfig, HpkeConfigId, HpkeKdfId, HpkeKemId, - HpkePublicKey, Interval, TaskId, Time, TimePrecision, Url as DapUrl, + HpkePublicKey, Interval, TaskId, Time, TimePrecision, Url as DapUrl, VdafConfig, }; use rand::random; use serde_json::json; @@ -1552,6 +1655,151 @@ mod tests { ); } + #[test] + fn own_endpoint_is_this_aggregators_endpoint() { + // Each aggregator's own endpoint is its own side of the pair (not the peer's); a swap here + // would produce a mismatched TaskConfiguration and break every AAD decryption. + let task = TaskBuilder::new( + BatchMode::TimeInterval, + AggregationMode::Synchronous, + VdafInstance::Prio3Count, + ) + .with_leader_aggregator_endpoint("https://leader.example.com/".parse().unwrap()) + .with_helper_aggregator_endpoint("https://helper.example.com/".parse().unwrap()) + .build(); + + let leader = task.leader_view().unwrap(); + assert_eq!( + leader.own_aggregator_endpoint().as_str(), + "https://leader.example.com/" + ); + assert_eq!( + leader.peer_aggregator_endpoint().as_str(), + "https://helper.example.com/" + ); + + let helper = task.helper_view().unwrap(); + assert_eq!( + helper.own_aggregator_endpoint().as_str(), + "https://helper.example.com/" + ); + assert_eq!( + helper.peer_aggregator_endpoint().as_str(), + "https://leader.example.com/" + ); + } + + #[test] + fn own_endpoint_stored_verbatim() { + // The production constructor must not re-encode the own endpoint: a non-canonical value + // (mixed-case host, no trailing slash) is bound verbatim into the task's TaskConfiguration + // (DAP-18 §4.1). A `url::Url`-normalizing path would silently canonicalize it. + let noncanonical = "https://Example.COM/DAP"; + let task = TaskBuilder::new( + BatchMode::TimeInterval, + AggregationMode::Synchronous, + VdafInstance::Prio3Count, + ) + .build() + .leader_view() + .unwrap(); + let task = AggregatorTask::new( + *task.id(), + task.peer_aggregator_endpoint().clone(), + noncanonical.try_into().unwrap(), + *task.batch_mode(), + task.vdaf().clone(), + task.opaque_vdaf_verify_key().clone(), + task.task_interval().copied(), + task.report_expiry_age().copied(), + task.min_batch_size(), + *task.time_precision(), + *task.tolerable_clock_skew(), + task.task_info().to_vec(), + task.aggregator_parameters().clone(), + ) + .unwrap(); + assert_eq!(task.own_aggregator_endpoint().as_str(), noncanonical); + } + + #[test] + fn task_configuration_synthesized_for_api_task() { + // For an API-provisioned task, task_configuration() synthesizes from stored parameters, + // pairing own/peer endpoints into leader/helper by role. Both aggregators' views must + // produce byte-identical configurations, or their AADs would not match. + let task = TaskBuilder::new( + BatchMode::TimeInterval, + AggregationMode::Synchronous, + VdafInstance::Prio3Count, + ) + .with_leader_aggregator_endpoint("https://leader.example.com/".parse().unwrap()) + .with_helper_aggregator_endpoint("https://helper.example.com/".parse().unwrap()) + .build(); + + let leader_config = task.leader_view().unwrap().task_configuration().unwrap(); + assert_eq!( + leader_config.leader_aggregator_endpoint().as_str(), + "https://leader.example.com/" + ); + assert_eq!( + leader_config.helper_aggregator_endpoint().as_str(), + "https://helper.example.com/" + ); + assert_eq!( + leader_config, + task.helper_view().unwrap().task_configuration().unwrap() + ); + } + + #[test] + fn task_configuration_verbatim_for_taskprov() { + // A taskprov task binds the received wire configuration verbatim, not one synthesized from + // its stored parameters. Prove this by giving the wire config a distinct task_info: if the + // task re-synthesized, it would use the task's own task_info instead. + let wire = build_task_configuration( + b"wire-specific-task-info".to_vec(), + "https://leader.example.com/".try_into().unwrap(), + "https://helper.example.com/".try_into().unwrap(), + TimePrecision::from_seconds(3600), + 100, + BatchConfig::TimeInterval, + VdafConfig::Prio3Count, + None, + ) + .unwrap(); + + let task = TaskBuilder::new( + BatchMode::TimeInterval, + AggregationMode::Synchronous, + VdafInstance::Prio3Count, + ) + .build() + .taskprov_helper_view() + .unwrap() + .with_taskprov_task_config(wire.clone()); + + assert_eq!(task.task_configuration().unwrap(), wire); + assert_eq!( + task.task_configuration().unwrap().task_info(), + b"wire-specific-task-info" + ); + } + + #[test] + fn task_configuration_taskprov_without_config_errors() { + // A taskprov task whose verbatim wire config is absent must fail loudly rather than + // silently synthesize a (non-byte-faithful) config from its stored parameters. + let task = TaskBuilder::new( + BatchMode::TimeInterval, + AggregationMode::Synchronous, + VdafInstance::Prio3Count, + ) + .build() + .taskprov_helper_view() + .unwrap(); + assert_matches!(task.task_configuration(), Err(Error::InvalidParameter(_))); + } + #[test] fn aggregator_request_paths() { for (prefix, task) in [ @@ -1645,6 +1893,7 @@ mod tests { AggregatorTask::new( TaskId::from([0; 32]), peer_aggregator_endpoint, + "https://example.com/".parse().unwrap(), BatchMode::TimeInterval, VdafInstance::Prio3Count, SecretBytes::new(b"1234567812345678".to_vec()), @@ -1709,13 +1958,15 @@ mod tests { &[ Token::Struct { name: "SerializedAggregatorTask", - len: 18, + len: 19, }, Token::Str("task_id"), Token::Some, Token::Str("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"), Token::Str("peer_aggregator_endpoint"), Token::Str("https://example.net/"), + Token::Str("own_aggregator_endpoint"), + Token::Str("https://example.com/"), Token::Str("batch_mode"), Token::UnitVariant { name: "BatchMode", @@ -1820,6 +2071,7 @@ mod tests { &AggregatorTask::new( TaskId::from([255; 32]), "https://example.com/".parse().unwrap(), + "https://example.net/".parse().unwrap(), BatchMode::LeaderSelected { batch_time_window_size: None, }, @@ -1863,13 +2115,15 @@ mod tests { &[ Token::Struct { name: "SerializedAggregatorTask", - len: 18, + len: 19, }, Token::Str("task_id"), Token::Some, Token::Str("__________________________________________8"), Token::Str("peer_aggregator_endpoint"), Token::Str("https://example.com/"), + Token::Str("own_aggregator_endpoint"), + Token::Str("https://example.net/"), Token::Str("batch_mode"), Token::StructVariant { name: "BatchMode", diff --git a/client/src/lib.rs b/client/src/lib.rs index fc7a56483..919cab7e1 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -8,7 +8,7 @@ //! //! ```no_run //! use prio::vdaf::prio3::Prio3Histogram; -//! use janus_messages::{TimePrecision, TaskId, Url}; +//! use janus_messages::{BatchConfig, TimePrecision, TaskId, Url, VdafConfig}; //! use std::str::FromStr; //! //! #[tokio::main] @@ -23,13 +23,20 @@ //! let taskid = "rc0jgm1MHH6Q7fcI4ZdNUxas9DAYLcJFK5CL7xUl-gU"; //! let task = TaskId::from_str(taskid).unwrap(); //! -//! let client = janus_client::Client::new( +//! // The task parameters below are bound into HPKE AADs and MUST match those provisioned to the +//! // aggregators byte-for-byte. +//! let client = janus_client::Client::builder( //! task, //! leader_url, //! helper_url, //! TimePrecision::from_seconds(300), -//! vdaf +//! vdaf, //! ) +//! .with_task_info(b"[task info]".to_vec()) +//! .with_min_batch_size(1000) +//! .with_batch_config(BatchConfig::TimeInterval) +//! .with_vdaf_config(VdafConfig::Prio3Histogram { length: 12, chunk_length: 4 }) +//! .build() //! .await //! .unwrap(); //! client.upload(5).await.unwrap(); @@ -56,14 +63,16 @@ use janus_core::{ retries::{ ExponentialWithTotalDelayBuilder, http_request_exponential_backoff, retry_http_request, }, + task_config::build_task_configuration, time::{Clock, DateTimeExt, RealClock}, url_for_join, vdaf::vdaf_application_context, }; use janus_messages::{ - HpkeConfig, HpkeConfigList, InputShareAad, MediaType, PlaintextInputShare, Report, ReportId, - ReportMetadata, ReportUploadStatus, Role, TaskId, Time, TimePrecision, UploadErrors, - UploadRequest, Url as DapUrl, + BatchConfig, HpkeConfig, HpkeConfigList, InputShareAad, Interval, MediaType, + PlaintextInputShare, Report, ReportId, ReportMetadata, ReportUploadStatus, Role, + TaskConfiguration, TaskId, Time, TimePrecision, UploadErrors, UploadRequest, Url as DapUrl, + VdafConfig, }; #[cfg(feature = "ohttp")] use ohttp::{ClientRequest, KeyConfig}; @@ -122,6 +131,8 @@ pub enum Error { #[cfg(feature = "ohttp")] #[error("No supported key configurations advertised by OHTTP gateway")] OhttpNoSupportedKeyConfigs(Box>), + #[error(transparent)] + Message(#[from] janus_messages::Error), } impl From for Error { @@ -169,6 +180,20 @@ struct ClientParameters { /// The time precision of the task. This value is shared by all parties in the protocol, and is /// used to compute report timestamps. time_precision: TimePrecision, + /// Opaque task info bound into the task's `TaskConfiguration`. Required before uploading; set + /// via [`ClientBuilder::with_task_info`]. + task_info: Option>, + /// Minimum batch size, bound into the task's `TaskConfiguration`. Required before uploading; + /// set via [`ClientBuilder::with_min_batch_size`]. + min_batch_size: Option, + /// Batch configuration bound into the task's `TaskConfiguration`. Required before uploading; + /// set via [`ClientBuilder::with_batch_config`]. + batch_config: Option, + /// VDAF configuration bound into the task's `TaskConfiguration`. Required before uploading; + /// set via [`ClientBuilder::with_vdaf_config`]. + vdaf_config: Option, + /// Optional task validity interval bound into the task's `TaskConfiguration`. + task_interval: Option, /// Parameters to use when retrying HTTP requests. http_request_retry_parameters: ExponentialWithTotalDelayBuilder, } @@ -188,10 +213,38 @@ impl ClientParameters { leader_aggregator_endpoint, helper_aggregator_endpoint, time_precision, + task_info: None, + min_batch_size: None, + batch_config: None, + vdaf_config: None, + task_interval: None, http_request_retry_parameters: http_request_exponential_backoff(), } } + /// Builds this task's canonical [`TaskConfiguration`] for binding into HPKE AADs. Errors if any + /// of the required parameters (`task_info`, `min_batch_size`, `batch_config`, `vdaf_config`) + /// were not supplied via the corresponding [`ClientBuilder`] setters. + fn task_configuration(&self) -> Result { + Ok(build_task_configuration( + self.task_info + .clone() + .ok_or(Error::InvalidParameter("task_info not set"))?, + self.leader_aggregator_endpoint.clone(), + self.helper_aggregator_endpoint.clone(), + self.time_precision, + self.min_batch_size + .ok_or(Error::InvalidParameter("min_batch_size not set"))?, + self.batch_config + .clone() + .ok_or(Error::InvalidParameter("batch_config not set"))?, + self.vdaf_config + .clone() + .ok_or(Error::InvalidParameter("vdaf_config not set"))?, + self.task_interval, + )?) + } + /// The URL relative to which the API endpoints for the aggregator may be found, if the role is /// an aggregator, or an error otherwise. fn aggregator_endpoint(&self, role: &Role) -> Result<&DapUrl, Error> { @@ -282,6 +335,10 @@ impl> ClientBuilder { /// Finalize construction of a [`Client`]. This will fetch HPKE configurations from each /// aggregator via HTTPS. pub async fn build(self) -> Result, Error> { + // Fail fast if any required TaskConfiguration parameter is unset, rather than deferring to + // the first upload's AAD construction. + self.parameters.task_configuration()?; + let http_client = if let Some(http_client) = self.http_client { http_client } else { @@ -333,6 +390,9 @@ impl> ClientBuilder { leader_hpke_config: HpkeConfig, helper_hpke_config: HpkeConfig, ) -> Result, Error> { + // Fail fast on unset params (see `build`). + self.parameters.task_configuration()?; + let http_client = if let Some(http_client) = self.http_client { http_client } else { @@ -368,6 +428,40 @@ impl> ClientBuilder { self } + /// Set the opaque task info bound into the task's `TaskConfiguration`. Required before + /// [`Self::build`]. + pub fn with_task_info(mut self, task_info: Vec) -> Self { + self.parameters.task_info = Some(task_info); + self + } + + /// Set the minimum batch size bound into the task's `TaskConfiguration`. Required before + /// [`Self::build`]. + pub fn with_min_batch_size(mut self, min_batch_size: u64) -> Self { + self.parameters.min_batch_size = Some(min_batch_size); + self + } + + /// Set the batch configuration bound into the task's `TaskConfiguration`. Required before + /// [`Self::build`]. + pub fn with_batch_config(mut self, batch_config: BatchConfig) -> Self { + self.parameters.batch_config = Some(batch_config); + self + } + + /// Set the VDAF configuration bound into the task's `TaskConfiguration`. Required before + /// [`Self::build`]. + pub fn with_vdaf_config(mut self, vdaf_config: VdafConfig) -> Self { + self.parameters.vdaf_config = Some(vdaf_config); + self + } + + /// Set the optional task validity interval bound into the task's `TaskConfiguration`. + pub fn with_task_interval(mut self, task_interval: Option) -> Self { + self.parameters.task_interval = task_interval; + self + } + /// Set the leader HPKE configuration to be used, preventing the client from fetching it from /// the aggregator over HTTPS. pub fn with_leader_hpke_config(mut self, hpke_config: HpkeConfig) -> Self { @@ -437,55 +531,6 @@ pub struct Client> { } impl> Client { - /// Construct a new client from the required set of DAP task parameters. - pub async fn new( - task_id: TaskId, - leader_aggregator_endpoint: DapUrl, - helper_aggregator_endpoint: DapUrl, - time_precision: TimePrecision, - vdaf: V, - ) -> Result { - ClientBuilder::new( - task_id, - leader_aggregator_endpoint, - helper_aggregator_endpoint, - time_precision, - vdaf, - ) - .build() - .await - } - - /// Construct a new client, and provide the aggregator HPKE configurations through an - /// out-of-band means. - /// - /// # Notes - /// - /// This method is not compatible with OHTTP. Use [`ClientBuilder::with_ohttp_config`] and then - /// [`ClientBuilder::build`] to provide OHTTP configuration. - #[deprecated( - note = "Use `ClientBuilder::with_leader_hpke_config`, `ClientBuilder::with_helper_hpke_config` and `ClientBuilder::build` instead" - )] - pub fn with_hpke_configs( - task_id: TaskId, - leader_aggregator_endpoint: DapUrl, - helper_aggregator_endpoint: DapUrl, - time_precision: TimePrecision, - vdaf: V, - leader_hpke_config: HpkeConfig, - helper_hpke_config: HpkeConfig, - ) -> Result { - #[allow(deprecated)] - ClientBuilder::new( - task_id, - leader_aggregator_endpoint, - helper_aggregator_endpoint, - time_precision, - vdaf, - ) - .build_with_hpke_configs(leader_hpke_config, helper_hpke_config) - } - /// Creates a [`ClientBuilder`] for further configuration from the required set of DAP task /// parameters. pub fn builder( @@ -584,7 +629,7 @@ impl> Client { /// /// ```no_run /// # use janus_client::{Client, Error}; - /// # use janus_messages::{TimePrecision, Time}; + /// # use janus_messages::{BatchConfig, TimePrecision, Time, VdafConfig}; /// # use prio::vdaf::prio3::Prio3; /// # use rand::random; /// # use std::time::SystemTime; @@ -594,13 +639,18 @@ impl> Client { /// # let measurement2 = false; /// # let vdaf = Prio3::new_count(2).unwrap(); /// let time_precision = TimePrecision::from_seconds(3600); - /// let client = Client::new( + /// let client = Client::builder( /// random(), /// "https://example.com/".parse().unwrap(), /// "https://example.net/".parse().unwrap(), /// time_precision, /// vdaf, - /// ).await?; + /// ) + /// .with_task_info(b"[task info]".to_vec()) + /// .with_min_batch_size(1000) + /// .with_batch_config(BatchConfig::TimeInterval) + /// .with_vdaf_config(VdafConfig::Prio3Count) + /// .build().await?; /// /// // Upload multiple measurements with explicit timestamps. /// // Can use SystemTime for wall clock times: @@ -868,20 +918,25 @@ where /// /// ```no_run /// # use janus_client::{Client, UploadStats}; - /// # use janus_messages::{TimePrecision, Time}; + /// # use janus_messages::{BatchConfig, TimePrecision, Time, VdafConfig}; /// # use prio::vdaf::prio3::Prio3Count; /// # use rand::random; /// # /// # #[tokio::main] /// # async fn main() { /// let time_precision = TimePrecision::from_seconds(300); - /// let client = Client::new( + /// let client = Client::builder( /// random(), /// "https://leader.example.com/".parse().unwrap(), /// "https://helper.example.com/".parse().unwrap(), /// time_precision, /// Prio3Count::new_count(2).unwrap(), - /// ).await.unwrap(); + /// ) + /// .with_task_info(b"[task info]".to_vec()) + /// .with_min_batch_size(1000) + /// .with_batch_config(BatchConfig::TimeInterval) + /// .with_vdaf_config(VdafConfig::Prio3Count) + /// .build().await.unwrap(); /// /// let session = client.upload_session(100); /// let now = Time::from_seconds_since_epoch(1_000_000, &time_precision); diff --git a/client/src/tests/mod.rs b/client/src/tests/mod.rs index 22470926c..0860058ce 100644 --- a/client/src/tests/mod.rs +++ b/client/src/tests/mod.rs @@ -7,8 +7,8 @@ use janus_core::{ test_util::install_test_trace_subscriber, }; use janus_messages::{ - HpkeConfigList, MediaType, ReportError, ReportUploadStatus, Role, Time, TimePrecision, - UploadErrors, UploadRequest, Url as DapUrl, + BatchConfig, HpkeConfigList, MediaType, ReportError, ReportUploadStatus, Role, Time, + TimePrecision, UploadErrors, UploadRequest, Url as DapUrl, VdafConfig, }; use prio::{ codec::Encode, @@ -31,6 +31,10 @@ async fn setup_client>(server: &mockito::Server, vdaf: V) -> vdaf, ) .with_backoff(test_http_request_exponential_backoff()) + .with_task_info(b"test task".to_vec()) + .with_min_batch_size(1) + .with_batch_config(BatchConfig::TimeInterval) + .with_vdaf_config(VdafConfig::Prio3Count) .with_leader_hpke_config(HpkeKeypair::test().config().clone()) .with_helper_hpke_config(HpkeKeypair::test().config().clone()) .build() @@ -38,6 +42,27 @@ async fn setup_client>(server: &mockito::Server, vdaf: V) -> .unwrap() } +#[tokio::test] +async fn build_fails_without_required_task_configuration() { + initialize_rustls(); + // Omitting the required TaskConfiguration setters must fail fast at build(), before any network + // access, rather than deferring to the first upload's AAD construction. + let server_url = DapUrl::try_from("https://example.com/").unwrap(); + let err = Client::builder( + random(), + server_url.clone(), + server_url, + TimePrecision::from_seconds(100), + Prio3::new_count(2).unwrap(), + ) + .with_leader_hpke_config(HpkeKeypair::test().config().clone()) + .with_helper_hpke_config(HpkeKeypair::test().config().clone()) + .build() + .await + .unwrap_err(); + assert_matches!(err, Error::InvalidParameter(_)); +} + #[test] fn endpoints_preserved_and_joined() { let task_id = random(); diff --git a/client/src/tests/ohttp.rs b/client/src/tests/ohttp.rs index 41dad8b6d..c67b1f378 100644 --- a/client/src/tests/ohttp.rs +++ b/client/src/tests/ohttp.rs @@ -10,7 +10,9 @@ use janus_core::{ retries::test_util::test_http_request_exponential_backoff, test_util::install_test_trace_subscriber, }; -use janus_messages::{MediaType, Report, TimePrecision, UploadRequest, Url as DapUrl}; +use janus_messages::{ + BatchConfig, MediaType, Report, TimePrecision, UploadRequest, Url as DapUrl, VdafConfig, +}; use ohttp::{ KeyConfig, SymmetricSuite, hpke::{Aead, Kdf}, @@ -41,6 +43,10 @@ async fn build_client(server: &mockito::ServerGuard) -> Result), + #[error("invalid parameter {0}")] + InvalidParameter(&'static str), } impl From for Error { @@ -349,6 +365,23 @@ pub struct CollectorBuilder { vdaf: V, /// The task's time precision. time_precision: TimePrecision, + /// The base URL of the helper's aggregator API endpoints. Required before [`Self::build`]; set + /// via [`Self::with_helper_endpoint`]. + helper_endpoint: Option, + /// Opaque task info bound into the task's `TaskConfiguration`. Required before + /// [`Self::build`]; set via [`Self::with_task_info`]. + task_info: Option>, + /// Minimum batch size bound into the task's `TaskConfiguration`. Required before + /// [`Self::build`]; set via [`Self::with_min_batch_size`]. + min_batch_size: Option, + /// Batch configuration bound into the task's `TaskConfiguration`. Required before + /// [`Self::build`]; set via [`Self::with_batch_config`]. + batch_config: Option, + /// VDAF configuration bound into the task's `TaskConfiguration`. Required before + /// [`Self::build`]; set via [`Self::with_vdaf_config`]. + vdaf_config: Option, + /// Optional task validity interval bound into the task's `TaskConfiguration`. + task_interval: Option, /// HTTPS client. http_client: Option, @@ -376,6 +409,12 @@ impl CollectorBuilder { hpke_keypair, vdaf, time_precision, + helper_endpoint: None, + task_info: None, + min_batch_size: None, + batch_config: None, + vdaf_config: None, + task_interval: None, http_client: None, http_request_retry_parameters: http_request_exponential_backoff(), collect_poll_wait_parameters: ExponentialWithTotalDelayBuilder::new() @@ -392,19 +431,39 @@ impl CollectorBuilder { } else { default_http_client()? }; - Ok(Collector { + let collector = Collector { task_id: self.task_id, // Stored verbatim; the trailing slash is applied at join time via `url_for_join`, so // the bytes bound into HPKE AADs stay exactly as provisioned (DAP §4.1). leader_endpoint: self.leader_endpoint, + helper_endpoint: self + .helper_endpoint + .ok_or(Error::InvalidParameter("helper_endpoint not set"))?, authentication: self.authentication, hpke_keypair: self.hpke_keypair, vdaf: self.vdaf, time_precision: self.time_precision, + task_info: self + .task_info + .ok_or(Error::InvalidParameter("task_info not set"))?, + min_batch_size: self + .min_batch_size + .ok_or(Error::InvalidParameter("min_batch_size not set"))?, + batch_config: self + .batch_config + .ok_or(Error::InvalidParameter("batch_config not set"))?, + vdaf_config: self + .vdaf_config + .ok_or(Error::InvalidParameter("vdaf_config not set"))?, + task_interval: self.task_interval, http_client, http_request_retry_parameters: self.http_request_retry_parameters, collect_poll_wait_parameters: self.collect_poll_wait_parameters, - }) + }; + // Fail fast if the resolved parameters cannot form a valid TaskConfiguration (e.g. invalid + // endpoint bytes), rather than deferring to the first collection's AAD construction. + collector.task_configuration()?; + Ok(collector) } /// Provide an HTTPS client for the collector. @@ -424,6 +483,47 @@ impl CollectorBuilder { self.collect_poll_wait_parameters = backoff; self } + + /// Set the base URL of the helper's aggregator API endpoints, bound into the task's + /// `TaskConfiguration`. Required before [`Self::build`]. + pub fn with_helper_endpoint(mut self, helper_endpoint: DapUrl) -> Self { + self.helper_endpoint = Some(helper_endpoint); + self + } + + /// Set the opaque task info bound into the task's `TaskConfiguration`. Required before + /// [`Self::build`]. + pub fn with_task_info(mut self, task_info: Vec) -> Self { + self.task_info = Some(task_info); + self + } + + /// Set the minimum batch size bound into the task's `TaskConfiguration`. Required before + /// [`Self::build`]. + pub fn with_min_batch_size(mut self, min_batch_size: u64) -> Self { + self.min_batch_size = Some(min_batch_size); + self + } + + /// Set the batch configuration bound into the task's `TaskConfiguration`. Required before + /// [`Self::build`]. + pub fn with_batch_config(mut self, batch_config: BatchConfig) -> Self { + self.batch_config = Some(batch_config); + self + } + + /// Set the VDAF configuration bound into the task's `TaskConfiguration`. Required before + /// [`Self::build`]. + pub fn with_vdaf_config(mut self, vdaf_config: VdafConfig) -> Self { + self.vdaf_config = Some(vdaf_config); + self + } + + /// Set the optional task validity interval bound into the task's `TaskConfiguration`. + pub fn with_task_interval(mut self, task_interval: Option) -> Self { + self.task_interval = task_interval; + self + } } /// A DAP collector. @@ -435,6 +535,9 @@ pub struct Collector { /// The base URL of the leader's aggregator API endpoints. #[educe(Debug(method(std::fmt::Display::fmt)))] leader_endpoint: DapUrl, + /// The base URL of the helper's aggregator API endpoints. + #[educe(Debug(method(std::fmt::Display::fmt)))] + helper_endpoint: DapUrl, /// The authentication information needed to communicate with the leader aggregator. authentication: AuthenticationToken, /// HPKE keypair used for decryption of aggregate shares. @@ -444,6 +547,17 @@ pub struct Collector { vdaf: V, /// The task's time precision. time_precision: TimePrecision, + /// Opaque task info bound into the task's `TaskConfiguration`. + #[educe(Debug(ignore))] + task_info: Vec, + /// Minimum batch size bound into the task's `TaskConfiguration`. + min_batch_size: u64, + /// Batch configuration bound into the task's `TaskConfiguration`. + batch_config: BatchConfig, + /// VDAF configuration bound into the task's `TaskConfiguration`. + vdaf_config: VdafConfig, + /// Optional task validity interval bound into the task's `TaskConfiguration`. + task_interval: Option, /// HTTPS client. #[educe(Debug(ignore))] @@ -455,27 +569,6 @@ pub struct Collector { } impl Collector { - /// Construct a new collector. This requires certain DAP task parameters and an - /// implementation of the task's VDAF. - pub fn new( - task_id: TaskId, - leader_endpoint: DapUrl, - authentication: AuthenticationToken, - hpke_keypair: HpkeKeypair, - vdaf: V, - time_precision: TimePrecision, - ) -> Result, Error> { - Self::builder( - task_id, - leader_endpoint, - authentication, - hpke_keypair, - vdaf, - time_precision, - ) - .build() - } - /// Construct a [`CollectorBuilder`] from required DAP task parameters and an implementation of /// the task's VDAF. pub fn builder( @@ -496,6 +589,20 @@ impl Collector { ) } + /// Builds this task's canonical [`TaskConfiguration`] for binding into aggregate-share AADs. + fn task_configuration(&self) -> Result { + Ok(build_task_configuration( + self.task_info.clone(), + self.leader_endpoint.clone(), + self.helper_endpoint.clone(), + self.time_precision, + self.min_batch_size, + self.batch_config.clone(), + self.vdaf_config.clone(), + self.task_interval, + )?) + } + /// Construct a URI for a collection. fn collection_job_uri(&self, collection_job_id: CollectionJobId) -> Result { Ok(url_for_join(&self.leader_endpoint)?.join(&format!( @@ -826,9 +933,9 @@ mod tests { test_util::{VdafTranscript, install_test_trace_subscriber, run_vdaf}, }; use janus_messages::{ - AggregateShareAad, BatchId, BatchSelector, CollectionJobId, CollectionJobReq, + AggregateShareAad, BatchConfig, BatchId, BatchSelector, CollectionJobId, CollectionJobReq, CollectionJobResp, Duration, HpkeCiphertext, Interval, MediaType, PartialBatchSelector, - Query, Role, TaskId, Time, TimePrecision, Url as DapUrl, + Query, Role, TaskId, Time, TimePrecision, Url as DapUrl, VdafConfig, batch_mode::{LeaderSelected, TimeInterval}, problem_type::DapProblemType, }; @@ -854,12 +961,17 @@ mod tests { let hpke_keypair = HpkeKeypair::test(); Collector::builder( random(), - server_url, + server_url.clone(), AuthenticationToken::new_bearer_token_from_string("Y29sbGVjdG9yIHRva2Vu").unwrap(), hpke_keypair, vdaf, TEST_TIME_PRECISION, ) + .with_helper_endpoint(server_url) + .with_task_info(b"test task".to_vec()) + .with_min_batch_size(1) + .with_batch_config(BatchConfig::TimeInterval) + .with_vdaf_config(VdafConfig::Prio3Count) .with_http_request_backoff(test_http_request_exponential_backoff()) .with_collect_poll_backoff(test_http_request_exponential_backoff()) .build() @@ -957,7 +1069,7 @@ mod tests { ("http://example.com/dap", "http://example.com/dap/"), ("http://example.com", "http://example.com/"), ] { - let collector = Collector::new( + let collector = Collector::builder( random(), endpoint.try_into().unwrap(), AuthenticationToken::new_bearer_token_from_string("Y29sbGVjdG9yIHRva2Vu").unwrap(), @@ -965,6 +1077,12 @@ mod tests { dummy::Vdaf::new(1), TEST_TIME_PRECISION, ) + .with_helper_endpoint("http://helper.example.com".try_into().unwrap()) + .with_task_info(b"test task".to_vec()) + .with_min_batch_size(1) + .with_batch_config(BatchConfig::TimeInterval) + .with_vdaf_config(VdafConfig::Prio3Count) + .build() .unwrap(); // Stored verbatim (no trailing slash added) for byte-identical HPKE AADs (DAP §4.1), @@ -1302,12 +1420,17 @@ mod tests { let hpke_keypair = HpkeKeypair::test(); let collector = Collector::builder( random(), - server_url, + server_url.clone(), AuthenticationToken::new_bearer_token_from_bytes(Vec::from([0x41u8; 16])).unwrap(), hpke_keypair, vdaf, TEST_TIME_PRECISION, ) + .with_helper_endpoint(server_url) + .with_task_info(b"test task".to_vec()) + .with_min_batch_size(1) + .with_batch_config(BatchConfig::TimeInterval) + .with_vdaf_config(VdafConfig::Prio3Count) .with_http_request_backoff(test_http_request_exponential_backoff()) .with_collect_poll_backoff(test_http_request_exponential_backoff()) .build() diff --git a/core/src/http/cached_resource.rs b/core/src/http/cached_resource.rs index f9ced99e9..07bdab094 100644 --- a/core/src/http/cached_resource.rs +++ b/core/src/http/cached_resource.rs @@ -188,20 +188,17 @@ pub(crate) fn expires_at<'a, I: IntoIterator>( return Some(Instant::now()); } - if let Some(max_age) = directive.strip_prefix("max-age=") { - let parsed = match max_age.parse() { - Ok(parsed) => parsed, - Err(_) => return None, - }; - - if expires_at.is_some() { - return None; - } - - expires_at = Instant::now().checked_add(Duration::from_secs(parsed)); - } else { + let max_age = directive.strip_prefix("max-age=")?; + let parsed = match max_age.parse() { + Ok(parsed) => parsed, + Err(_) => return None, + }; + + if expires_at.is_some() { return None; } + + expires_at = Instant::now().checked_add(Duration::from_secs(parsed)); } expires_at diff --git a/db/00000000000001_initial_schema.up.sql b/db/00000000000001_initial_schema.up.sql index 927d5544b..ca8b3ed9f 100644 --- a/db/00000000000001_initial_schema.up.sql +++ b/db/00000000000001_initial_schema.up.sql @@ -104,6 +104,8 @@ CREATE TABLE tasks( task_id BYTEA UNIQUE NOT NULL, -- 32-byte TaskID as defined by the DAP specification aggregator_role AGGREGATOR_ROLE NOT NULL, -- the role of this aggregator for this task peer_aggregator_endpoint TEXT NOT NULL, -- peer aggregator's API endpoint + own_aggregator_endpoint TEXT NOT NULL, -- this aggregator's own API endpoint + taskprov_task_config BYTEA, -- verbatim wire TaskConfiguration for taskprov tasks (bound into HPKE AADs); NULL for API-provisioned tasks batch_mode JSONB NOT NULL, -- the batch mode in use for this task, along with its parameters aggregation_mode AGGREGATION_MODE, -- the aggregation mode in use for this task (populated for Helper only) vdaf JSON NOT NULL, -- the VDAF instance in use for this task, along with its parameters diff --git a/docs/samples/tasks.yaml b/docs/samples/tasks.yaml index 4629e021b..a2b96c1be 100644 --- a/docs/samples/tasks.yaml +++ b/docs/samples/tasks.yaml @@ -9,6 +9,9 @@ # HTTPS endpoint of the peer aggregator. peer_aggregator_endpoint: "https://example.com/" + # HTTPS endpoint of this aggregator. + own_aggregator_endpoint: "https://example.net/" + # The DAP batch mode. See below for an example of a leader-selected task. batch_mode: TimeInterval @@ -102,6 +105,7 @@ - task_id: "D-hCKPuqL2oTf7ZVRVyMP5VGt43EAEA8q34mDf6p1JE" peer_aggregator_endpoint: "https://example.org/" + own_aggregator_endpoint: "https://example.net/" batch_mode: !LeaderSelected batch_time_window_size: null aggregation_mode: Synchronous diff --git a/integration_tests/src/client.rs b/integration_tests/src/client.rs index 992901e42..415888260 100644 --- a/integration_tests/src/client.rs +++ b/integration_tests/src/client.rs @@ -275,7 +275,17 @@ where helper_aggregator_endpoint.as_str().try_into().unwrap(), task_parameters.time_precision, vdaf, - ); + ) + .with_task_info(task_parameters.task_info.clone()) + .with_min_batch_size(task_parameters.min_batch_size) + .with_batch_config(task_parameters.batch_mode.to_batch_config()) + .with_vdaf_config( + task_parameters + .vdaf + .to_vdaf_config() + .map_err(janus_client::Error::InvalidParameter)?, + ) + .with_task_interval(task_parameters.task_interval); if let Some(ohttp_config) = &task_parameters.endpoint_fragments.ohttp_config { builder = builder.with_ohttp_config(ohttp_config.clone()); diff --git a/integration_tests/src/janus.rs b/integration_tests/src/janus.rs index 9cf2a88fc..4e9c72cff 100644 --- a/integration_tests/src/janus.rs +++ b/integration_tests/src/janus.rs @@ -24,9 +24,14 @@ use janus_aggregator::{ metrics::{MetricsConfiguration, test_util::InMemoryMetricInfrastructure}, trace::{TokioConsoleConfiguration, TraceConfiguration}, }; +#[cfg(feature = "testcontainer")] +use janus_aggregator_core::task::test_util::Task; use janus_aggregator_core::{ - datastore::test_util::{EphemeralDatastore, ephemeral_datastore}, - task::test_util::Task, + datastore::{ + Datastore, + test_util::{EphemeralDatastore, ephemeral_datastore}, + }, + task::AggregatorTask, }; use janus_core::time::RealClock; #[cfg(feature = "testcontainer")] @@ -34,6 +39,7 @@ use janus_interop_binaries::{ ContainerLogsDropGuard, get_rust_log_level, test_util::await_http_server, testcontainer::Aggregator, }; +#[cfg(feature = "testcontainer")] use janus_messages::Role; #[cfg(feature = "testcontainer")] use testcontainers::{ContainerRequest, ImageExt, runners::AsyncRunner}; @@ -96,6 +102,7 @@ impl JanusContainer { pub struct JanusInProcess { socket_address: SocketAddr, stopper: Stopper, + datastore: Datastore, _ephemeral_datastore: EphemeralDatastore, pub aggregator_metrics: InMemoryMetricInfrastructure, pub aggregation_job_creator_metrics: InMemoryMetricInfrastructure, @@ -105,9 +112,10 @@ pub struct JanusInProcess { } impl JanusInProcess { - /// Start a new Janus instance in the current process, using a separate ephemeral database, - /// configured to service the given task. - pub async fn new(task: &Task, role: Role) -> Self { + /// Start a new Janus instance in the current process, using a separate ephemeral database. The + /// task to service is provisioned separately via [`Self::provision_task`] once the aggregator's + /// ephemeral port is known. + pub async fn new() -> Self { // Set up common utilities. let stopper = Stopper::new(); let clock = RealClock::default(); @@ -126,12 +134,6 @@ impl JanusInProcess { let collection_job_driver_metrics = InMemoryMetricInfrastructure::new(); let key_rotator_metrics = InMemoryMetricInfrastructure::new(); - // Provision the task. - datastore - .put_aggregator_task(&task.view_for_role(role).expect("invalid role")) - .await - .expect("task provisioning failed"); - // Construct configuration and options for each component. let common_binary_options = CommonBinaryOptions { datastore_keys, @@ -332,9 +334,13 @@ impl JanusInProcess { .expect("aggregator task shut down before sending socket address"); }; + // A dedicated datastore handle for provisioning the task after startup. + let datastore = ephemeral_datastore.datastore(clock).await; + Self { socket_address, stopper, + datastore, _ephemeral_datastore: ephemeral_datastore, aggregator_metrics, aggregation_job_creator_metrics, @@ -348,6 +354,14 @@ impl JanusInProcess { pub fn port(&self) -> u16 { self.socket_address.port() } + + /// Provision the given aggregator task view into this instance's datastore. + pub async fn provision_task(&self, task: &AggregatorTask) { + self.datastore + .put_aggregator_task(task) + .await + .expect("task provisioning failed"); + } } impl Drop for JanusInProcess { diff --git a/integration_tests/src/lib.rs b/integration_tests/src/lib.rs index 3fa05b6a6..76c57c876 100644 --- a/integration_tests/src/lib.rs +++ b/integration_tests/src/lib.rs @@ -6,7 +6,7 @@ use janus_aggregator_core::task::BatchMode; use janus_client::OhttpConfig; use janus_collector::AuthenticationToken; use janus_core::{hpke::HpkeKeypair, vdaf::VdafInstance}; -use janus_messages::{TaskId, TimePrecision}; +use janus_messages::{Interval, TaskId, TimePrecision}; use url::Url; pub mod client; @@ -22,6 +22,11 @@ pub struct TaskParameters { pub vdaf: VdafInstance, pub min_batch_size: u64, pub time_precision: TimePrecision, + /// The task's `task_info`, sourced from the provisioned task so the client and collector bind + /// a byte-identical `TaskConfiguration` into their HPKE AADs. + pub task_info: Vec, + /// The task's optional collection interval, for the same byte-identity reason. + pub task_interval: Option, pub collector_hpke_keypair: HpkeKeypair, pub collector_auth_token: AuthenticationToken, pub collector_max_interval: time::Duration, diff --git a/integration_tests/tests/integration/common.rs b/integration_tests/tests/integration/common.rs index 6beb6e5ea..f582d510a 100644 --- a/integration_tests/tests/integration/common.rs +++ b/integration_tests/tests/integration/common.rs @@ -113,6 +113,8 @@ pub fn build_test_task( vdaf: task_builder.vdaf().clone(), min_batch_size: task_builder.min_batch_size(), time_precision: *task_builder.time_precision(), + task_info: task_builder.task_info().to_vec(), + task_interval: task_builder.task_interval().copied(), collector_hpke_keypair: task_builder.collector_hpke_keypair().clone(), collector_auth_token: task_builder.collector_auth_token().clone(), collector_max_interval, @@ -260,6 +262,13 @@ where vdaf, task_parameters.time_precision, ) + // The helper endpoint and task_info are threaded through for AAD byte-identity in a later + // stage; placeholders for now. + .with_helper_endpoint("http://unused.helper.example/".try_into().unwrap()) + .with_task_info(Vec::new()) + .with_min_batch_size(task_parameters.min_batch_size) + .with_batch_config(task_parameters.batch_mode.to_batch_config()) + .with_vdaf_config(task_parameters.vdaf.to_vdaf_config().unwrap()) .with_http_request_backoff(test_http_request_exponential_backoff()) .with_collect_poll_backoff( ExponentialWithTotalDelayBuilder::new() diff --git a/integration_tests/tests/integration/janus.rs b/integration_tests/tests/integration/janus.rs index 77d5af397..cab274f6a 100644 --- a/integration_tests/tests/integration/janus.rs +++ b/integration_tests/tests/integration/janus.rs @@ -10,6 +10,7 @@ use janus_integration_tests::janus::JanusContainer; use janus_integration_tests::{TaskParameters, client::ClientBackend, janus::JanusInProcess}; #[cfg(feature = "testcontainer")] use janus_interop_binaries::test_util::generate_network_name; +#[cfg(feature = "testcontainer")] use janus_messages::Role; use prio::{ dp::{ @@ -86,20 +87,36 @@ impl JanusInProcessPair { /// Set up a new pair of in-process Janus test instances, and set up a new task in each using /// the given VDAF and batch mode. pub async fn new(task_builder: TaskBuilder) -> JanusInProcessPair { - let (task_parameters, mut task_builder) = build_test_task( + let (task_parameters, task_builder) = build_test_task( task_builder, TestContext::Host, Duration::from_millis(500), Duration::from_secs(60), ); - let helper = JanusInProcess::new(&task_builder.clone().build(), Role::Helper).await; - let helper_url = task_parameters - .endpoint_fragments - .helper - .endpoint_for_host(helper.port()); - task_builder = task_builder.with_helper_aggregator_endpoint(helper_url); - let leader = JanusInProcess::new(&task_builder.build(), Role::Leader).await; + // Start both aggregators first, so their ephemeral ports are known. Then build a single + // task with byte-identical real endpoints and provision each aggregator's view of + // it. + let leader = JanusInProcess::new().await; + let helper = JanusInProcess::new().await; + + let task = task_builder + .with_leader_aggregator_endpoint( + task_parameters + .endpoint_fragments + .leader + .endpoint_for_host(leader.port()), + ) + .with_helper_aggregator_endpoint( + task_parameters + .endpoint_fragments + .helper + .endpoint_for_host(helper.port()), + ) + .build(); + + leader.provision_task(&task.leader_view().unwrap()).await; + helper.provision_task(&task.helper_view().unwrap()).await; Self { task_parameters, diff --git a/integration_tests/tests/integration/simulation/setup.rs b/integration_tests/tests/integration/simulation/setup.rs index adca6f5c8..c6ba02e4f 100644 --- a/integration_tests/tests/integration/simulation/setup.rs +++ b/integration_tests/tests/integration/simulation/setup.rs @@ -231,6 +231,11 @@ impl Components { state.vdaf.clone(), ) .with_http_client(http_client.clone()) + .with_task_info(leader_task.task_info().to_vec()) + .with_min_batch_size(leader_task.min_batch_size()) + .with_batch_config(leader_task.batch_mode().to_batch_config()) + .with_vdaf_config(leader_task.vdaf().to_vdaf_config().unwrap()) + .with_task_interval(leader_task.task_interval().copied()) .build() .await .unwrap(); @@ -342,6 +347,17 @@ impl Components { state.vdaf.clone(), *task.time_precision(), ) + .with_helper_endpoint( + task.helper_aggregator_endpoint() + .as_str() + .try_into() + .unwrap(), + ) + .with_task_info(leader_task.task_info().to_vec()) + .with_min_batch_size(leader_task.min_batch_size()) + .with_batch_config(leader_task.batch_mode().to_batch_config()) + .with_vdaf_config(leader_task.vdaf().to_vdaf_config().unwrap()) + .with_task_interval(leader_task.task_interval().copied()) .with_http_request_backoff(http_request_exponential_backoff()) .with_collect_poll_backoff( ExponentialWithTotalDelayBuilder::new().with_total_delay(Some(StdDuration::ZERO)), diff --git a/interop_binaries/src/commands/janus_interop_aggregator.rs b/interop_binaries/src/commands/janus_interop_aggregator.rs index 154914923..bab52d7a9 100644 --- a/interop_binaries/src/commands/janus_interop_aggregator.rs +++ b/interop_binaries/src/commands/janus_interop_aggregator.rs @@ -62,13 +62,18 @@ async fn handle_add_task( datastore: &Datastore, request: AggregatorAddTaskRequest, ) -> anyhow::Result<()> { - let peer_aggregator_endpoint: DapUrl = match request.role { - AggregatorRole::Leader => request.helper, - AggregatorRole::Helper => request.leader, - } - .as_str() - .parse() - .context("invalid peer aggregator endpoint")?; + let (peer_endpoint, own_endpoint) = match request.role { + AggregatorRole::Leader => (&request.helper, &request.leader), + AggregatorRole::Helper => (&request.leader, &request.helper), + }; + let peer_aggregator_endpoint: DapUrl = peer_endpoint + .as_str() + .parse() + .context("invalid peer aggregator endpoint")?; + let own_aggregator_endpoint: DapUrl = own_endpoint + .as_str() + .parse() + .context("invalid own aggregator endpoint")?; let vdaf = request.vdaf.into(); let leader_authentication_token = AuthenticationToken::new_dap_auth_token_from_string(request.leader_authentication_token) @@ -147,6 +152,7 @@ async fn handle_add_task( let task = AggregatorTask::new( request.task_id, peer_aggregator_endpoint, + own_aggregator_endpoint, batch_mode, vdaf, vdaf_verify_key, diff --git a/interop_binaries/src/commands/janus_interop_client.rs b/interop_binaries/src/commands/janus_interop_client.rs index fb36eb730..f72ef41c2 100644 --- a/interop_binaries/src/commands/janus_interop_client.rs +++ b/interop_binaries/src/commands/janus_interop_client.rs @@ -16,7 +16,7 @@ use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD}; use clap::Parser; use educe::Educe; use janus_core::vdaf::{VdafInstance, new_prio3_sum_vec_field64_multiproof_hmacsha256_aes128}; -use janus_messages::{TaskId, Time, TimePrecision, Url as DapUrl}; +use janus_messages::{BatchConfig, TaskId, Time, TimePrecision, Url as DapUrl}; use prio::{ codec::Decode, field::Field64, @@ -88,6 +88,10 @@ async fn handle_upload_generic>( let task_id = TaskId::get_decoded(&task_id_bytes).context("invalid length of TaskId")?; let time_precision = TimePrecision::from_seconds(request.time_precision); + let vdaf_config = VdafInstance::from(request.vdaf.clone()) + .to_vdaf_config() + .map_err(|err| anyhow::anyhow!("unsupported VDAF for TaskConfiguration: {err}"))?; + let client = janus_client::Client::builder( task_id, request.leader, @@ -96,6 +100,13 @@ async fn handle_upload_generic>( vdaf, ) .with_http_client(http_client.clone()) + // The interop upload request does not yet carry task_info, min_batch_size, batch_config, or + // task_interval; these placeholders make the client build, but interop AAD byte-identity needs + // test-design support to provide the real values. + .with_task_info(Vec::new()) + .with_min_batch_size(0) + .with_batch_config(BatchConfig::TimeInterval) + .with_vdaf_config(vdaf_config) .build() .await .context("failed to construct client")?; diff --git a/interop_binaries/src/commands/janus_interop_collector.rs b/interop_binaries/src/commands/janus_interop_collector.rs index da45fb3a6..4516e75c3 100644 --- a/interop_binaries/src/commands/janus_interop_collector.rs +++ b/interop_binaries/src/commands/janus_interop_collector.rs @@ -23,8 +23,8 @@ use janus_core::{ vdaf::{VdafInstance, new_prio3_sum_vec_field64_multiproof_hmacsha256_aes128}, }; use janus_messages::{ - BatchId, Duration, HpkeConfig, Interval, PartialBatchSelector, Query, TaskId, Time, - TimePrecision, Url as DapUrl, batch_mode::BatchMode, + BatchConfig, BatchId, Duration, HpkeConfig, Interval, PartialBatchSelector, Query, TaskId, + Time, TimePrecision, Url as DapUrl, batch_mode::BatchMode, }; use prio::{ codec::{Decode, Encode}, @@ -206,6 +206,9 @@ where B: BatchMode, { let time_precision = task_state.time_precision; + let vdaf_config = VdafInstance::from(task_state.vdaf.clone()) + .to_vdaf_config() + .map_err(|err| anyhow::anyhow!("unsupported VDAF for TaskConfiguration: {err}"))?; let collector = Collector::builder( task_state.task_id, task_state.leader_url.clone(), @@ -214,6 +217,15 @@ where vdaf, time_precision, ) + // The interop add-task request does not yet carry the helper endpoint, task_info, + // min_batch_size, batch_config, or task_interval; these placeholders make the collector + // build, but interop AAD byte-identity needs test-design support to provide the real + // values. + .with_helper_endpoint("http://unused.helper.example/".try_into().unwrap()) + .with_task_info(Vec::new()) + .with_min_batch_size(0) + .with_batch_config(BatchConfig::TimeInterval) + .with_vdaf_config(vdaf_config) .with_http_client(http_client.clone()) .with_http_request_backoff( ExponentialWithTotalDelayBuilder::new() diff --git a/messages/src/lib.rs b/messages/src/lib.rs index fb1d7bf3d..fbba0efc6 100644 --- a/messages/src/lib.rs +++ b/messages/src/lib.rs @@ -3071,7 +3071,7 @@ pub(crate) fn roundtrip_encoding_parameterized<'a, T, P, TB, PB, I>( impl Debug for Wrapper { fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { - write!(f, "{:02x?}", &self.0) + write!(f, "{:02x?}", self.0) } } diff --git a/tools/src/bin/collect.rs b/tools/src/bin/collect.rs index d329e07da..e52ebb50d 100644 --- a/tools/src/bin/collect.rs +++ b/tools/src/bin/collect.rs @@ -16,8 +16,8 @@ use janus_core::{ retries::ExponentialWithTotalDelayBuilder, }; use janus_messages::{ - CollectionJobId, Duration, HpkeConfig, Interval, PartialBatchSelector, Query, TaskId, Time, - TimePrecision, Url as DapUrl, + BatchConfig, CollectionJobId, Duration, HpkeConfig, Interval, PartialBatchSelector, Query, + TaskId, Time, TimePrecision, Url as DapUrl, VdafConfig, batch_mode::{BatchMode, LeaderSelected, TimeInterval}, }; use prio::{ @@ -623,6 +623,14 @@ fn new_collector( vdaf, time_precision, ) + // The helper endpoint and remaining TaskConfiguration parameters are not yet exposed as CLI + // options; these placeholders make the collector build, but correct aggregate-share decryption + // will require real values once the TaskConfiguration is bound into the AAD. + .with_helper_endpoint("http://unused.helper.example/".try_into().unwrap()) + .with_task_info(Vec::new()) + .with_min_batch_size(0) + .with_batch_config(BatchConfig::TimeInterval) + .with_vdaf_config(VdafConfig::Prio3Count) .with_http_client(http_client) .with_collect_poll_backoff( ExponentialWithTotalDelayBuilder::new()