From 5e28bd7dcec563906f09517608ceb2f382d8b12f Mon Sep 17 00:00:00 2001 From: "J.C. Jones" Date: Fri, 10 Jul 2026 16:46:19 -0700 Subject: [PATCH 1/7] fix lints for clippy 1.97 (#4708) --- aggregator_api/src/tests.rs | 18 +++++++++--------- core/src/http/cached_resource.rs | 21 +++++++++------------ messages/src/lib.rs | 2 +- 3 files changed, 19 insertions(+), 22 deletions(-) diff --git a/aggregator_api/src/tests.rs b/aggregator_api/src/tests.rs index f86336160..7137b79f8 100644 --- a/aggregator_api/src/tests.rs +++ b/aggregator_api/src/tests.rs @@ -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) 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/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) } } From dbecf85953a28f51e0ac5d3308569ddb246ae3d3 Mon Sep 17 00:00:00 2001 From: "J.C. Jones" Date: Fri, 10 Jul 2026 11:27:51 -0700 Subject: [PATCH 2/7] [DAP-18] TaskConf AAD: Provision in-process test task after aggregator startup Restructure JanusInProcess to provision its task via a new provision_task() method once its ephemeral port is known, so JanusInProcessPair builds one task with consistent real endpoints and provisions both aggregator views. Doesn't do anything else. --- integration_tests/src/janus.rs | 36 ++++++++++++++------ integration_tests/tests/integration/janus.rs | 33 +++++++++++++----- 2 files changed, 50 insertions(+), 19 deletions(-) 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/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, From a972ec2d61279df0430a23b34713e30d4b95c716 Mon Sep 17 00:00:00 2001 From: "J.C. Jones" Date: Tue, 7 Jul 2026 15:35:03 -0700 Subject: [PATCH 3/7] [DAP-17] TaskConf AAD Part 3: persist each aggregator's own endpoint DAP-18 binds the task's TaskConfiguration -- which contains both the leader and helper endpoints -- into the HPKE AADs for input and aggregate shares. AggregatorTask previously stored only peer_aggregator_endpoint, so a party could not reconstruct both endpoints from the task alone. Add an own_aggregator_endpoint field (and DB column) alongside the peer endpoint: - API-provisioned tasks stamp it at creation from the aggregator's configured public_dap_url. - taskprov tasks populate it from the received TaskConfiguration's endpoint for this aggregator's role, which taskprov_opt_in previously discarded. In the process, we move to the janus_messages::Url. --- aggregator/src/aggregator.rs | 5 +- aggregator/src/binaries/aggregator.rs | 4 +- aggregator/src/binaries/janus_cli.rs | 2 + aggregator_api/src/lib.rs | 5 +- aggregator_api/src/models.rs | 2 +- aggregator_api/src/routes.rs | 1 + aggregator_api/src/tests.rs | 1 + aggregator_core/src/datastore.rs | 13 ++- aggregator_core/src/datastore/tests.rs | 48 ++++++++++ aggregator_core/src/task.rs | 96 ++++++++++++++++++- db/00000000000001_initial_schema.up.sql | 1 + docs/samples/tasks.yaml | 4 + .../src/commands/janus_interop_aggregator.rs | 20 ++-- 13 files changed, 183 insertions(+), 19 deletions(-) diff --git a/aggregator/src/aggregator.rs b/aggregator/src/aggregator.rs index 9ed4036a9..01644f9e2 100644 --- a/aggregator/src/aggregator.rs +++ b/aggregator/src/aggregator.rs @@ -849,7 +849,9 @@ 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. + let (peer_aggregator, leader_url, helper_url) = self .taskprov_authorize_request(peer_role, task_id, task_config, aggregator_auth_token) .await?; @@ -874,6 +876,7 @@ impl Aggregator { AggregatorTask::new( *task_id, leader_url, + helper_url, BatchMode::try_from(task_config.batch_config())?, vdaf_instance, vdaf_verify_key, 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 7137b79f8..857ca1052 100644 --- a/aggregator_api/src/tests.rs +++ b/aggregator_api/src/tests.rs @@ -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..9ffd90e48 100644 --- a/aggregator_core/src/datastore.rs +++ b/aggregator_core/src/datastore.rs @@ -713,7 +713,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, 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 +722,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 ) ON CONFLICT DO NOTHING", ) @@ -735,6 +736,8 @@ 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(), /* batch_mode */ &Json(task.batch_mode()), /* vdaf */ &Json(task.vdaf()), /* task_start */ @@ -889,7 +892,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, 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 +917,7 @@ FROM tasks WHERE task_id = $1", "-- get_aggregator_tasks() SELECT task_id, aggregator_role, aggregation_mode, peer_aggregator_endpoint, + own_aggregator_endpoint, 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 +939,7 @@ 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 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 +1054,7 @@ FROM tasks", let task = AggregatorTask::new( *task_id, peer_aggregator_endpoint, + own_aggregator_endpoint, batch_mode, vdaf, vdaf_verify_key, diff --git a/aggregator_core/src/datastore/tests.rs b/aggregator_core/src/datastore/tests.rs index 6aed85589..58daafcf8 100644 --- a/aggregator_core/src/datastore/tests.rs +++ b/aggregator_core/src/datastore/tests.rs @@ -341,6 +341,54 @@ 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 set_task_deactivate_at(ephemeral_datastore: EphemeralDatastore) { diff --git a/aggregator_core/src/task.rs b/aggregator_core/src/task.rs index e683e476f..91932e02e 100644 --- a/aggregator_core/src/task.rs +++ b/aggregator_core/src/task.rs @@ -254,6 +254,9 @@ 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 @@ -267,6 +270,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 +297,7 @@ impl AggregatorTask { Self::new_with_common_parameters( common_parameters, peer_aggregator_endpoint, + own_aggregator_endpoint, aggregator_parameters, ) } @@ -300,11 +305,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,6 +333,7 @@ impl AggregatorTask { Ok(Self { common_parameters, peer_aggregator_endpoint, + own_aggregator_endpoint, aggregator_parameters, deactivate_at: None, }) @@ -351,6 +359,11 @@ 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 + } + /// Retrieves the batch mode associated with this task. pub fn batch_mode(&self) -> &BatchMode { &self.common_parameters.batch_mode @@ -683,6 +696,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 +757,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 +852,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 +1154,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 +1170,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 +1186,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, }, @@ -1552,6 +1571,73 @@ 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 aggregator_request_paths() { for (prefix, task) in [ @@ -1645,6 +1731,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 +1796,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 +1909,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 +1953,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/db/00000000000001_initial_schema.up.sql b/db/00000000000001_initial_schema.up.sql index 927d5544b..a65581298 100644 --- a/db/00000000000001_initial_schema.up.sql +++ b/db/00000000000001_initial_schema.up.sql @@ -104,6 +104,7 @@ 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 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/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, From 652436b334d5e1420fed0418ffbf6e37ab8bc552 Mon Sep 17 00:00:00 2001 From: "J.C. Jones" Date: Tue, 7 Jul 2026 16:07:28 -0700 Subject: [PATCH 4/7] [DAP-17] TaskConf AAD Part 3: add AggregatorTask::task_configuration() Add the adapter that yields the canonical TaskConfiguration bound into HPKE AADs, so the upcoming AAD sites can obtain it from the task alone. For API-provisioned tasks it is synthesized via build_task_configuration, pairing the task's own and peer endpoints into leader/helper by role. For taskprov tasks it is the received wire configuration verbatim. Store that configuration on the task (new nullable taskprov_task_config column, populated in taskprov_opt_in). Re-synthesizing it from the stored parameters is not byte-safe: the VdafConfig<->VdafInstance round trip drops DP strategies and unknown task extensions would be lost, and any divergence breaks decryption. --- aggregator/src/aggregator.rs | 6 +- aggregator/src/aggregator/taskprov_tests.rs | 24 ++- aggregator_core/src/datastore.rs | 26 +++- aggregator_core/src/datastore/tests.rs | 34 +++++ aggregator_core/src/task.rs | 158 +++++++++++++++++++- db/00000000000001_initial_schema.up.sql | 1 + 6 files changed, 236 insertions(+), 13 deletions(-) diff --git a/aggregator/src/aggregator.rs b/aggregator/src/aggregator.rs index 01644f9e2..e3606e334 100644 --- a/aggregator/src/aggregator.rs +++ b/aggregator/src/aggregator.rs @@ -851,6 +851,7 @@ impl Aggregator { ) -> Result<(), Error> { // 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?; @@ -897,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_core/src/datastore.rs b/aggregator_core/src/datastore.rs index 9ffd90e48..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,8 +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, - own_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, @@ -722,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, $25 + $19, $20, $21, $22, $23, $24, $25, $26 ) ON CONFLICT DO NOTHING", ) @@ -738,6 +739,11 @@ ON CONFLICT DO NOTHING", &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 */ @@ -893,7 +899,7 @@ UPDATE tasks SET "-- get_aggregator_task() SELECT aggregator_role, aggregation_mode, peer_aggregator_endpoint, - own_aggregator_endpoint, batch_mode, + 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, @@ -917,7 +923,7 @@ FROM tasks WHERE task_id = $1", "-- get_aggregator_tasks() SELECT task_id, aggregator_role, aggregation_mode, peer_aggregator_endpoint, - own_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, @@ -940,6 +946,10 @@ FROM tasks", 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; @@ -1067,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 58daafcf8..81b16f2b8 100644 --- a/aggregator_core/src/datastore/tests.rs +++ b/aggregator_core/src/datastore/tests.rs @@ -389,6 +389,40 @@ async fn roundtrip_task_own_endpoint_verbatim(ephemeral_datastore: EphemeralData 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 91932e02e..6f6a1e315 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 @@ -262,6 +265,14 @@ pub struct AggregatorTask { /// 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 { @@ -336,6 +347,7 @@ impl AggregatorTask { own_aggregator_endpoint, aggregator_parameters, deactivate_at: None, + taskprov_task_config: None, }) } @@ -364,6 +376,61 @@ impl AggregatorTask { &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 @@ -405,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() @@ -1470,12 +1543,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; @@ -1638,6 +1712,84 @@ mod tests { 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 [ diff --git a/db/00000000000001_initial_schema.up.sql b/db/00000000000001_initial_schema.up.sql index a65581298..ca8b3ed9f 100644 --- a/db/00000000000001_initial_schema.up.sql +++ b/db/00000000000001_initial_schema.up.sql @@ -105,6 +105,7 @@ CREATE TABLE tasks( 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 From 095c802cdd7b6d99bff8e6eda2a68663b8f167d4 Mon Sep 17 00:00:00 2001 From: "J.C. Jones" Date: Wed, 8 Jul 2026 16:50:47 -0700 Subject: [PATCH 5/7] [DAP-17] TaskConf AAD Part 3: expand Client with TaskConfiguration params Add builder setters (with_task_info, with_min_batch_size, with_batch_config, with_vdaf_config, with_task_interval) supplying the parameters the client needs to reconstruct the task's canonical TaskConfiguration for HPKE AADs. The four non-optional parameters are validated at build(), which fails fast rather than deferring to the first upload's AAD construction. ClientParameters::task_configuration() is the single construction path; it is not yet AAD-load-bearing (wired in the flag day). In-tree builder callers (tests, integration harness, interop client) are updated to supply the fields. --- client/src/lib.rs | 177 ++++++++++++------ client/src/tests/mod.rs | 29 ++- client/src/tests/ohttp.rs | 8 +- integration_tests/src/client.rs | 10 + .../tests/integration/simulation/setup.rs | 5 + .../src/commands/janus_interop_client.rs | 13 +- 6 files changed, 177 insertions(+), 65 deletions(-) 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>( 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")?; From 3449c377373d818b682f68b6e344e36fa4476983 Mon Sep 17 00:00:00 2001 From: "J.C. Jones" Date: Wed, 8 Jul 2026 19:40:36 -0700 Subject: [PATCH 6/7] [DAP-17] TaskConf AAD Part 3: expand Collector with TaskConfiguration params Add builder setters (with_helper_endpoint, with_task_info, with_min_batch_size, with_batch_config, with_vdaf_config, with_task_interval) supplying the parameters the collector needs to reconstruct the task's canonical TaskConfiguration for aggregate-share AADs. The collector had no helper endpoint before. Required fields are validated at build(), which also fails fast on invalid endpoint bytes rather than deferring to the first collection's AAD construction. Collector::task_configuration() is the single construction path; it is not yet AAD-load-bearing (wired in the flag day). Collector::new is removed (it bypassed the setters and so could no longer build). In-tree callers (tests, integration harness, interop collector, collect tool) are updated to supply the fields. --- collector/src/lib.rs | 189 +++++++++++++++--- integration_tests/tests/integration/common.rs | 7 + .../tests/integration/simulation/setup.rs | 11 + .../src/commands/janus_interop_collector.rs | 16 +- tools/src/bin/collect.rs | 12 +- 5 files changed, 198 insertions(+), 37 deletions(-) diff --git a/collector/src/lib.rs b/collector/src/lib.rs index f7dc36f32..40ec502ed 100644 --- a/collector/src/lib.rs +++ b/collector/src/lib.rs @@ -12,7 +12,9 @@ //! use std::{fs::File, str::FromStr}; //! //! use janus_collector::{Collector, PrivateCollectorCredential}; -//! use janus_messages::{Duration, Interval, Query, TaskId, Time, TimePrecision, Url}; +//! use janus_messages::{ +//! BatchConfig, Duration, Interval, Query, TaskId, Time, TimePrecision, Url, VdafConfig, +//! }; //! use prio::vdaf::prio3::Prio3; //! //! # async fn run() { @@ -27,11 +29,16 @@ //! let leader_url = //! Url::from_str("[absolute URI to the DAP leader, e.g. https://leader.dap.example.com/]") //! .unwrap(); +//! let helper_url = +//! Url::from_str("[absolute URI to the DAP helper, e.g. https://helper.dap.example.com/]") +//! .unwrap(); //! //! // Supply a VDAF implementation, corresponding to this task. //! let vdaf = Prio3::new_count(2).unwrap(); //! -//! let collector = Collector::new( +//! // The task parameters below are bound into HPKE AADs and MUST match those provisioned to the +//! // aggregators byte-for-byte. +//! let collector = Collector::builder( //! task_id, //! leader_url, //! collector_credential.authentication_token(), @@ -39,6 +46,12 @@ //! vdaf, //! time_precision, //! ) +//! .with_helper_endpoint(helper_url) +//! .with_task_info(b"[task info]".to_vec()) +//! .with_min_batch_size(1000) +//! .with_batch_config(BatchConfig::TimeInterval) +//! .with_vdaf_config(VdafConfig::Prio3Count) +//! .build() //! .unwrap(); //! //! // If this is a time interval task, specify the time interval over which the aggregation @@ -82,13 +95,14 @@ use janus_core::{ retries::{ ExponentialWithTotalDelayBuilder, http_request_exponential_backoff, retry_http_request, }, + task_config::build_task_configuration, time::TimeExt, url_for_join, }; use janus_messages::{ - AggregateShareAad, BatchSelector, CollectionJobExtension, CollectionJobId, CollectionJobReq, - CollectionJobResp, MediaType, PartialBatchSelector, Query, Role, TaskId, TimePrecision, - Url as DapUrl, + AggregateShareAad, BatchConfig, BatchSelector, CollectionJobExtension, CollectionJobId, + CollectionJobReq, CollectionJobResp, Interval, MediaType, PartialBatchSelector, Query, Role, + TaskConfiguration, TaskId, TimePrecision, Url as DapUrl, VdafConfig, batch_mode::{BatchMode, TimeInterval}, }; use prio::{ @@ -139,6 +153,8 @@ pub enum Error { Message(#[from] janus_messages::Error), #[error("the response from the server was invalid: {0}")] BadResponse(Box), + #[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/integration_tests/tests/integration/common.rs b/integration_tests/tests/integration/common.rs index 6beb6e5ea..0a9b49640 100644 --- a/integration_tests/tests/integration/common.rs +++ b/integration_tests/tests/integration/common.rs @@ -260,6 +260,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/simulation/setup.rs b/integration_tests/tests/integration/simulation/setup.rs index 803990e02..c6ba02e4f 100644 --- a/integration_tests/tests/integration/simulation/setup.rs +++ b/integration_tests/tests/integration/simulation/setup.rs @@ -347,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_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/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() From b72429c29040273981ff97d36a0741d011c59e00 Mon Sep 17 00:00:00 2001 From: "J.C. Jones" Date: Fri, 10 Jul 2026 15:26:34 -0700 Subject: [PATCH 7/7] Thread byte-identical TaskConfiguration values through the test harness Source task_info and the task interval from the provisioned task and thread them into the client (and TaskParameters) so tests build a TaskConfiguration that matches the aggregators'. Add TaskBuilder task_info()/task_interval() getters. Values are threaded but not yet cryptographically bound (the AAD struct changes land separately), so this is behavior-preserving. Also fix the get_config expectation to the byte-preserving dap_url (no trailing slash), which follows from persisting the own endpoint verbatim. --- aggregator_api/src/tests.rs | 2 +- aggregator_core/src/task.rs | 10 ++++++++++ integration_tests/src/client.rs | 6 +++--- integration_tests/src/lib.rs | 7 ++++++- integration_tests/tests/integration/common.rs | 2 ++ 5 files changed, 22 insertions(+), 5 deletions(-) diff --git a/aggregator_api/src/tests.rs b/aggregator_api/src/tests.rs index 857ca1052..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","#, diff --git a/aggregator_core/src/task.rs b/aggregator_core/src/task.rs index 6f6a1e315..9cc55c903 100644 --- a/aggregator_core/src/task.rs +++ b/aggregator_core/src/task.rs @@ -1515,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() diff --git a/integration_tests/src/client.rs b/integration_tests/src/client.rs index a6c6f2d2b..415888260 100644 --- a/integration_tests/src/client.rs +++ b/integration_tests/src/client.rs @@ -276,8 +276,7 @@ where task_parameters.time_precision, vdaf, ) - // task_info is threaded through for AAD byte-identity in a later stage; empty for now. - .with_task_info(Vec::new()) + .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( @@ -285,7 +284,8 @@ where .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/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 0a9b49640..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,