Repository: cloudflare/daphne
Pull request #728 (2024-11-27 15:27:52)

Make more use of DapRole enums
The DapRole enum is underused and under featured, leading to code like `is_leader: bool` where an instance of the enum would make calling code more readable (there are places where `false /* helper */` can be see where `DapRole::Helper` would be much better)

This PR also factors out the calculation of the `info` and `aad` parameters needed for hpke encryption and decryption. Making this part both more type safe and taking it out of higher level code

This PR should be reviewed one commit at a time


Patch:
diff --git a/crates/dapf/src/main.rs b/crates/dapf/src/main.rs
index e74afea..31f7316 100644
--- a/crates/dapf/src/main.rs
+++ b/crates/dapf/src/main.rs
@@ -14,6 +14,7 @@ use dapf::{
     HttpClient,
 };
 use daphne::{
+    constants::DapAggregatorRole,
     hpke::HpkeReceiverConfig,
     messages::{
         self, encode_base64url, BatchSelector, CollectionReq, PartialBatchSelector, Query, TaskId,
@@ -23,7 +24,6 @@ use daphne::{
 use daphne_service_utils::{
     bearer_token::BearerToken,
     test_route_types::{InternalTestAddTask, InternalTestVdaf},
-    DapRole,
 };
 use prio::codec::{ParameterizedDecode, ParameterizedEncode};
 use rand::{thread_rng, Rng};
@@ -236,7 +236,7 @@ enum TestAction {
         #[arg(long)]
         collector_auth_token: Option<String>,
         #[arg(long)]
-        role: Option<DapRole>,
+        role: Option<DapAggregatorRole>,
         #[arg(long)]
         query: Option<CliDapQueryConfig>,
         #[arg(long)]
@@ -866,11 +866,11 @@ async fn handle_test_routes(action: TestAction, http_client: HttpClient) -> anyh
                     "leader auth token",
                 )?,
                 collector_authentication_token: match role {
-                    DapRole::Leader => Some(use_or_request_from_user(
+                    DapAggregatorRole::Leader => Some(use_or_request_from_user(
                         collector_auth_token,
                         "collector auth token",
                     )?),
-                    DapRole::Helper => None,
+                    DapAggregatorRole::Helper => None,
                 },
                 role,
                 vdaf_verify_key,
diff --git a/crates/daphne-server/examples/service.rs b/crates/daphne-server/examples/service.rs
index 328d9e0..cd1f266 100644
--- a/crates/daphne-server/examples/service.rs
+++ b/crates/daphne-server/examples/service.rs
@@ -4,10 +4,10 @@
 use std::path::PathBuf;
 
 use clap::Parser;
+use daphne::constants::DapAggregatorRole;
 use daphne_server::{
     config::DaphneServiceConfig, metrics::DaphnePromServiceMetrics, router, App, StorageProxyConfig,
 };
-use daphne_service_utils::DapRole;
 use serde::{Deserialize, Serialize};
 use tokio::net::TcpListener;
 use tracing_subscriber::EnvFilter;
@@ -47,8 +47,8 @@ impl TryFrom<Args> for Config {
                     config::Value::new(
                         Some(&String::from("args.role")),
                         match role {
-                            DapRole::Leader => "leader",
-                            DapRole::Helper => "helper",
+                            DapAggregatorRole::Leader => "leader",
+                            DapAggregatorRole::Helper => "helper",
                         },
                     )
                 }),
@@ -81,7 +81,7 @@ struct Args {
     // --- command line overridable parameters ---
     /// One of `leader` or `helper`.
     #[arg(short, long)]
-    role: Option<DapRole>,
+    role: Option<DapAggregatorRole>,
     /// The port to listen on.
     #[arg(short, long)]
     port: Option<u16>,
diff --git a/crates/daphne-server/src/config.rs b/crates/daphne-server/src/config.rs
index bd0ace0..52c686d 100644
--- a/crates/daphne-server/src/config.rs
+++ b/crates/daphne-server/src/config.rs
@@ -2,10 +2,11 @@
 // SPDX-License-Identifier: BSD-3-Clause
 
 use daphne::{
+    constants::DapAggregatorRole,
     hpke::{HpkeConfig, HpkeReceiverConfig},
     DapGlobalConfig, DapVersion,
 };
-use daphne_service_utils::{bearer_token::BearerToken, DapRole};
+use daphne_service_utils::bearer_token::BearerToken;
 use p256::ecdsa::SigningKey;
 use serde::{Deserialize, Serialize};
 use url::Url;
@@ -45,7 +46,7 @@ pub type HpkeRecieverConfigList = Vec<HpkeReceiverConfig>;
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct DaphneServiceConfig {
     /// Indicates the role the service should play.
-    pub role: DapRole,
+    pub role: DapAggregatorRole,
 
     /// Global DAP configuration.
     #[serde(flatten)]
diff --git a/crates/daphne-server/src/lib.rs b/crates/daphne-server/src/lib.rs
index 102dfce..75cd4d4 100644
--- a/crates/daphne-server/src/lib.rs
+++ b/crates/daphne-server/src/lib.rs
@@ -6,10 +6,11 @@ use std::sync::Arc;
 use config::{DaphneServiceConfig, PeerBearerToken};
 use daphne::{
     audit_log::{AuditLog, NoopAuditLog},
+    constants::DapRole,
     fatal_error,
     messages::{Base64Encode, TaskId},
     roles::{leader::in_memory_leader::InMemoryLeaderState, DapAggregator},
-    DapError, DapSender,
+    DapError,
 };
 use daphne_service_utils::bearer_token::BearerToken;
 use either::Either::{self, Left, Right};
@@ -42,7 +43,7 @@ mod storage_proxy_connection;
 /// ```
 /// use std::num::NonZeroUsize;
 /// use url::Url;
-/// use daphne::{DapGlobalConfig, hpke::HpkeKemId, DapVersion};
+/// use daphne::{DapGlobalConfig, constants::DapAggregatorRole, hpke::HpkeKemId, DapVersion};
 /// use daphne_server::{
 ///     App,
 ///     router,
@@ -50,7 +51,6 @@ mod storage_proxy_connection;
 ///     metrics::DaphnePromServiceMetrics,
 ///     config::DaphneServiceConfig,
 /// };
-/// use daphne_service_utils::DapRole;
 ///
 /// let storage_proxy_settings = StorageProxyConfig {
 ///     url: Url::parse("http://example.com").unwrap(),
@@ -66,7 +66,7 @@ mod storage_proxy_connection;
 ///     default_num_agg_span_shards: NonZeroUsize::new(2).unwrap(),
 /// };
 /// let service_config = DaphneServiceConfig {
-///     role: DapRole::Helper,
+///     role: DapAggregatorRole::Helper,
 ///     global,
 ///     base_url: None,
 ///     taskprov: None,
@@ -77,7 +77,7 @@ mod storage_proxy_connection;
 /// };
 /// let app = App::new(storage_proxy_settings, daphne_service_metrics, service_config)?;
 ///
-/// let router = router::new(DapRole::Helper, app);
+/// let router = router::new(DapAggregatorRole::Helper, app);
 ///
 /// # Ok::<(), daphne::DapError>(())
 /// ```
@@ -114,7 +114,7 @@ impl router::DaphneService for App {
     async fn check_bearer_token(
         &self,
         presented_token: &BearerToken,
-        sender: DapSender,
+        sender: DapRole,
         task_id: TaskId,
         is_taskprov: bool,
     ) -> Result<(), Either<String, DapError>> {
@@ -131,16 +131,16 @@ impl router::DaphneService for App {
             .filter(|_| self.service_config.taskprov.is_some() && is_taskprov)
         {
             match (&taskprov.peer_auth, sender) {
-                (PeerBearerToken::Leader { expected_token }, DapSender::Leader)
-                | (PeerBearerToken::Collector { expected_token }, DapSender::Collector)
+                (PeerBearerToken::Leader { expected_token }, DapRole::Leader)
+                | (PeerBearerToken::Collector { expected_token }, DapRole::Collector)
                     if expected_token == presented_token =>
                 {
                     Ok(())
                 }
-                (PeerBearerToken::Leader { .. }, DapSender::Collector) => Err(Right(fatal_error!(
+                (PeerBearerToken::Leader { .. }, DapRole::Collector) => Err(Right(fatal_error!(
                     err = "expected a leader sender but got a collector sender"
                 ))),
-                (PeerBearerToken::Collector { .. }, DapSender::Leader) => Err(Right(fatal_error!(
+                (PeerBearerToken::Collector { .. }, DapRole::Leader) => Err(Right(fatal_error!(
                     err = "expected a collector sender but got a leader sender"
                 ))),
                 _ => reject(format_args!("using taskprov")),
diff --git a/crates/daphne-server/src/roles/aggregator.rs b/crates/daphne-server/src/roles/aggregator.rs
index e2b2564..f00e9d8 100644
--- a/crates/daphne-server/src/roles/aggregator.rs
+++ b/crates/daphne-server/src/roles/aggregator.rs
@@ -8,7 +8,7 @@ use daphne::{
     audit_log::AuditLog,
     error::DapAbort,
     fatal_error,
-    hpke::{self, HpkeConfig, HpkeProvider, HpkeReceiverConfig},
+    hpke::{self, info_and_aad, HpkeConfig, HpkeProvider, HpkeReceiverConfig},
     messages::{self, BatchId, BatchSelector, HpkeCiphertext, TaskId, Time},
     metrics::DaphneMetrics,
     roles::{
@@ -363,11 +363,10 @@ pub struct HpkeDecrypter(Marc<Vec<HpkeReceiverConfig>>);
 impl hpke::HpkeDecrypter for HpkeDecrypter {
     fn hpke_decrypt(
         &self,
-        info: &[u8],
-        aad: &[u8],
+        info: impl info_and_aad::InfoAndAad,
         ciphertext: &HpkeCiphertext,
     ) -> Result<Vec<u8>, DapError> {
-        self.0.hpke_decrypt(info, aad, ciphertext)
+        self.0.hpke_decrypt(info, ciphertext)
     }
 }
 
diff --git a/crates/daphne-server/src/roles/leader.rs b/crates/daphne-server/src/roles/leader.rs
index 3e0bcb1..dde6f62 100644
--- a/crates/daphne-server/src/roles/leader.rs
+++ b/crates/daphne-server/src/roles/leader.rs
@@ -5,7 +5,7 @@ use std::{borrow::Cow, time::Instant};
 
 use axum::{async_trait, http::Method};
 use daphne::{
-    constants::DapMediaType,
+    constants::{DapMediaType, DapRole},
     error::DapAbort,
     fatal_error,
     messages::{BatchId, BatchSelector, Collection, CollectionJobId, Report, TaskId},
@@ -171,7 +171,7 @@ impl crate::App {
             }
         } else if let Some(bearer_token) = self
             .bearer_tokens()
-            .get(daphne::DapSender::Leader, meta.task_id)
+            .get(DapRole::Leader, meta.task_id)
             .await
             .map_err(|e| fatal_error!(err = ?e, "failed to get leader bearer token"))?
         {
diff --git a/crates/daphne-server/src/roles/mod.rs b/crates/daphne-server/src/roles/mod.rs
index 2f95e55..397c2fa 100644
--- a/crates/daphne-server/src/roles/mod.rs
+++ b/crates/daphne-server/src/roles/mod.rs
@@ -1,7 +1,7 @@
 // Copyright (c) 2024 Cloudflare, Inc. All rights reserved.
 // SPDX-License-Identifier: BSD-3-Clause
 
-use daphne::{messages::TaskId, DapSender, ReplayProtection};
+use daphne::{constants::DapRole, messages::TaskId, ReplayProtection};
 use daphne_service_utils::bearer_token::BearerToken;
 
 use crate::storage_proxy_connection::{
@@ -54,12 +54,12 @@ impl BearerTokens<'_> {
     #[cfg(feature = "test-utils")]
     pub async fn put_if_not_exists(
         &self,
-        role: DapSender,
+        sender: DapRole,
         task_id: TaskId,
         token: BearerToken,
     ) -> Result<Option<BearerToken>, storage_proxy_connection::Error> {
         self.kv
-            .put_if_not_exists::<kv::prefix::KvBearerToken>(&(role, task_id).into(), token)
+            .put_if_not_exists::<kv::prefix::KvBearerToken>(&(sender, task_id).into(), token)
             .await
     }
 
@@ -72,13 +72,13 @@ impl BearerTokens<'_> {
     /// - `Err(error)` if any io error occurs while fetching
     pub async fn matches(
         &self,
-        role: DapSender,
+        sender: DapRole,
         task_id: TaskId,
         token: &BearerToken,
     ) -> Result<bool, Marc<storage_proxy_connection::Error>> {
         self.kv
             .peek::<kv::prefix::KvBearerToken, _, _>(
-                &(role, task_id).into(),
+                &(sender, task_id).into(),
                 &kv::KvGetOptions {
                     cache_not_found: false,
                 },
@@ -90,12 +90,12 @@ impl BearerTokens<'_> {
 
     pub async fn get(
         &self,
-        role: DapSender,
+        sender: DapRole,
         task_id: TaskId,
     ) -> Result<Option<BearerToken>, Marc<storage_proxy_connection::Error>> {
         self.kv
             .get_cloned::<kv::prefix::KvBearerToken>(
-                &(role, task_id).into(),
+                &(sender, task_id).into(),
                 &kv::KvGetOptions {
                     cache_not_found: false,
                 },
@@ -107,17 +107,17 @@ impl BearerTokens<'_> {
 #[cfg(feature = "test-utils")]
 mod test_utils {
     use daphne::{
+        constants::{DapAggregatorRole, DapRole},
         fatal_error,
         hpke::{HpkeConfig, HpkeReceiverConfig},
         messages::decode_base64url_vec,
         roles::DapAggregator,
         vdaf::{Prio3Config, VdafConfig},
-        DapError, DapQueryConfig, DapSender, DapTaskConfig, DapVersion,
+        DapError, DapQueryConfig, DapTaskConfig, DapVersion,
     };
     use daphne_service_utils::{
         bearer_token::BearerToken,
         test_route_types::{InternalTestAddTask, InternalTestEndpointForTask},
-        DapRole,
     };
     use prio::codec::Decode;
     use std::num::NonZeroUsize;
@@ -236,7 +236,7 @@ mod test_utils {
             let token = BearerToken::from(cmd.leader_authentication_token);
             if self
                 .bearer_tokens()
-                .put_if_not_exists(DapSender::Leader, cmd.task_id, token)
+                .put_if_not_exists(DapRole::Leader, cmd.task_id, token)
                 .await
                 .is_err()
             {
@@ -248,11 +248,11 @@ mod test_utils {
 
             // Collector authentication token.
             match (cmd.role, cmd.collector_authentication_token) {
-                (DapRole::Leader, Some(token_string)) => {
+                (DapAggregatorRole::Leader, Some(token_string)) => {
                     let token = BearerToken::from(token_string);
                     if self
                         .bearer_tokens()
-                        .put_if_not_exists(DapSender::Collector, cmd.task_id, token)
+                        .put_if_not_exists(DapRole::Collector, cmd.task_id, token)
                         .await
                         .is_err()
                     {
@@ -262,13 +262,13 @@ mod test_utils {
                         )));
                     }
                 }
-                (DapRole::Leader, None) => {
+                (DapAggregatorRole::Leader, None) => {
                     return Err(fatal_error!(
                         err = "command failed: missing collector authentication token",
                     ))
                 }
-                (DapRole::Helper, None) => (),
-                (DapRole::Helper, Some(..)) => {
+                (DapAggregatorRole::Helper, None) => (),
+                (DapAggregatorRole::Helper, Some(..)) => {
                     return Err(fatal_error!(
                         err = "command failed: unexpected collector authentication token",
                     ));
diff --git a/crates/daphne-server/src/router/extractor.rs b/crates/daphne-server/src/router/extractor.rs
index ef3b3ce..cc8d6b7 100644
--- a/crates/daphne-server/src/router/extractor.rs
+++ b/crates/daphne-server/src/router/extractor.rs
@@ -224,12 +224,12 @@ pub mod dap_sender {
     pub const FROM_HELPER: DapSender = 1 << 2;
     pub const FROM_LEADER: DapSender = 1 << 3;
 
-    pub const fn to_enum(id: DapSender) -> daphne::DapSender {
+    pub const fn to_enum(id: DapSender) -> daphne::constants::DapRole {
         match id {
-            FROM_CLIENT => daphne::DapSender::Client,
-            FROM_COLLECTOR => daphne::DapSender::Collector,
-            FROM_HELPER => daphne::DapSender::Helper,
-            FROM_LEADER => daphne::DapSender::Leader,
+            FROM_CLIENT => daphne::constants::DapRole::Client,
+            FROM_COLLECTOR => daphne::constants::DapRole::Collector,
+            FROM_HELPER => daphne::constants::DapRole::Helper,
+            FROM_LEADER => daphne::constants::DapRole::Leader,
             _ => panic!("invalid dap sender. Please specify a valid dap_sender from the crate::extractor::dap_sender module"),
         }
     }
@@ -344,14 +344,14 @@ mod test {
     };
     use daphne::{
         async_test_versions,
-        constants::DapMediaType,
+        constants::{DapMediaType, DapRole},
         messages::{
             request::{CollectionPollReq, RequestBody},
             taskprov::TaskprovAdvertisement,
             AggregationJobId, AggregationJobInitReq, Base64Encode, CollectionJobId, CollectionReq,
             TaskId,
         },
-        DapError, DapRequest, DapRequestMeta, DapSender, DapVersion,
+        DapError, DapRequest, DapRequestMeta, DapVersion,
     };
     use daphne_service_utils::{bearer_token::BearerToken, http_headers};
     use either::Either::{self, Left};
@@ -409,7 +409,7 @@ mod test {
             async fn check_bearer_token(
                 &self,
                 token: &BearerToken,
-                _sender: DapSender,
+                _sender: DapRole,
                 _task_id: TaskId,
                 _is_taskprov: bool,
             ) -> Result<(), Either<String, DapError>> {
diff --git a/crates/daphne-server/src/router/mod.rs b/crates/daphne-server/src/router/mod.rs
index 044ffba..7362d1d 100644
--- a/crates/daphne-server/src/router/mod.rs
+++ b/crates/daphne-server/src/router/mod.rs
@@ -18,10 +18,13 @@ use axum::{
     Json,
 };
 use daphne::{
-    error::DapAbort, fatal_error, messages::TaskId, DapError, DapRequestMeta, DapResponse,
-    DapSender,
+    constants::{DapAggregatorRole, DapRole},
+    error::DapAbort,
+    fatal_error,
+    messages::TaskId,
+    DapError, DapRequestMeta, DapResponse,
 };
-use daphne_service_utils::{bearer_token::BearerToken, DapRole};
+use daphne_service_utils::bearer_token::BearerToken;
 use either::Either;
 
 use crate::{metrics::DaphneServiceMetrics, App};
@@ -49,7 +52,7 @@ pub trait DaphneService {
     async fn check_bearer_token(
         &self,
         presented_token: &BearerToken,
-        sender: DapSender,
+        sender: DapRole,
         task_id: TaskId,
         is_taskprov: bool,
     ) -> Result<(), Either<String, DapError>>;
@@ -74,7 +77,7 @@ where
     async fn check_bearer_token(
         &self,
         presented_token: &BearerToken,
-        sender: DapSender,
+        sender: DapRole,
         task_id: TaskId,
         is_taskprov: bool,
     ) -> Result<(), Either<String, DapError>> {
@@ -86,14 +89,14 @@ where
     }
 }
 
-pub fn new(role: DapRole, aggregator: App) -> axum::Router<()> {
+pub fn new(role: DapAggregatorRole, aggregator: App) -> axum::Router<()> {
     let router = axum::Router::new();
 
     let router = aggregator::add_aggregator_routes(router);
 
     let router = match role {
-        DapRole::Leader => leader::add_leader_routes(router),
-        DapRole::Helper => helper::add_helper_routes(router),
+        DapAggregatorRole::Leader => leader::add_leader_routes(router),
+        DapAggregatorRole::Helper => helper::add_helper_routes(router),
     };
 
     #[cfg(feature = "test-utils")]
diff --git a/crates/daphne-server/src/storage_proxy_connection/kv/mod.rs b/crates/daphne-server/src/storage_proxy_connection/kv/mod.rs
index 9c74592..6552116 100644
--- a/crates/daphne-server/src/storage_proxy_connection/kv/mod.rs
+++ b/crates/daphne-server/src/storage_proxy_connection/kv/mod.rs
@@ -58,8 +58,9 @@ pub mod prefix {
     };
 
     use daphne::{
+        constants::DapRole,
         messages::{Base64Encode, TaskId},
-        taskprov, DapSender, DapTaskConfig, DapVersion,
+        taskprov, DapTaskConfig, DapVersion,
     };
     use daphne_service_utils::bearer_token::BearerToken;
     use serde::{de::DeserializeOwned, Serialize};
@@ -134,9 +135,9 @@ pub mod prefix {
     }
 
     #[derive(Debug)]
-    pub struct KvBearerTokenKey(DapSender, TaskId);
-    impl From<(DapSender, TaskId)> for KvBearerTokenKey {
-        fn from((s, t): (DapSender, TaskId)) -> Self {
+    pub struct KvBearerTokenKey(DapRole, TaskId);
+    impl From<(DapRole, TaskId)> for KvBearerTokenKey {
+        fn from((s, t): (DapRole, TaskId)) -> Self {
             Self(s, t)
         }
     }
@@ -145,10 +146,10 @@ pub mod prefix {
             let Self(sender, task_id) = self;
             let task_id = task_id.to_base64url();
             match sender {
-                DapSender::Client => write!(f, "client/task/{task_id}"),
-                DapSender::Collector => write!(f, "collector/task/{task_id}"),
-                DapSender::Helper => write!(f, "helper/task/{task_id}"),
-                DapSender::Leader => write!(f, "leader/task/{task_id}"),
+                DapRole::Client => write!(f, "client/task/{task_id}"),
+                DapRole::Collector => write!(f, "collector/task/{task_id}"),
+                DapRole::Helper => write!(f, "helper/task/{task_id}"),
+                DapRole::Leader => write!(f, "leader/task/{task_id}"),
             }
         }
     }
diff --git a/crates/daphne-service-utils/src/lib.rs b/crates/daphne-service-utils/src/lib.rs
index 66f71b1..3804665 100644
--- a/crates/daphne-service-utils/src/lib.rs
+++ b/crates/daphne-service-utils/src/lib.rs
@@ -3,11 +3,6 @@
 
 #![cfg_attr(not(test), deny(unused_crate_dependencies))]
 
-use core::fmt;
-use std::str::FromStr;
-
-use serde::{Deserialize, Serialize};
-
 pub mod bearer_token;
 #[cfg(feature = "durable_requests")]
 pub mod durable_requests;
@@ -25,41 +20,3 @@ mod durable_request_capnp {
         "/src/durable_requests/durable_request_capnp.rs"
     ));
 }
-
-#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
-#[serde(rename_all = "snake_case")]
-pub enum DapRole {
-    Leader,
-    Helper,
-}
-
-impl DapRole {
-    pub fn is_leader(self) -> bool {
-        self == Self::Leader
-    }
-
-    pub fn is_helper(self) -> bool {
-        self == Self::Helper
-    }
-}
-
-impl FromStr for DapRole {
-    type Err = String;
-
-    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
-        match s {
-            "leader" => Ok(Self::Leader),
-            "helper" => Ok(Self::Helper),
-            _ => Err(s.to_string()),
-        }
-    }
-}
-
-impl fmt::Display for DapRole {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.write_str(match self {
-            Self::Helper => "helper",
-            Self::Leader => "leader",
-        })
-    }
-}
diff --git a/crates/daphne/src/constants.rs b/crates/daphne/src/constants.rs
index 2ff93dc..dc709fe 100644
--- a/crates/daphne/src/constants.rs
+++ b/crates/daphne/src/constants.rs
@@ -4,6 +4,8 @@
 //! Constants used in the DAP protocol.
 
 use crate::DapVersion;
+use serde::{Deserialize, Serialize};
+use std::{fmt, str::FromStr};
 
 // Media types for HTTP requests.
 const MEDIA_TYPE_AGG_JOB_INIT_REQ: &str = "application/dap-aggregation-job-init-req";
@@ -63,6 +65,100 @@ impl DapMediaType {
     }
 }
 
+/// A role in the DAP aggregation protocol.
+#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum DapRole {
+    Collector,
+    Client,
+    Leader,
+    Helper,
+}
+
+impl DapRole {
+    pub fn is_leader(self) -> bool {
+        self == Self::Leader
+    }
+
+    pub fn is_helper(self) -> bool {
+        self == Self::Helper
+    }
+}
+
+impl FromStr for DapRole {
+    type Err = String;
+
+    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
+        match s {
+            "leader" => Ok(Self::Leader),
+            "helper" => Ok(Self::Helper),
+            "collector" => Ok(Self::Collector),
+            "client" => Ok(Self::Client),
+            _ => Err(s.to_string()),
+        }
+    }
+}
+
+impl fmt::Display for DapRole {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(match self {
+            Self::Client => "client",
+            Self::Collector => "collector",
+            Self::Helper => "helper",
+            Self::Leader => "leader",
+        })
+    }
+}
+
+/// A role in the DAP aggregation protocol. See [`DapRole`] for an explanation of the numeric values
+/// associated with each variant.
+#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "snake_case")]
+pub enum DapAggregatorRole {
+    Leader,
+    Helper,
+}
+
+impl DapAggregatorRole {
+    pub fn is_leader(self) -> bool {
+        self == Self::Leader
+    }
+
+    pub fn is_helper(self) -> bool {
+        self == Self::Helper
+    }
+}
+
+impl FromStr for DapAggregatorRole {
+    type Err = String;
+
+    fn from_str(s: &str) -> std::result::Result<Self, Self::Err> {
+        match s {
+            "leader" => Ok(Self::Leader),
+            "helper" => Ok(Self::Helper),
+            _ => Err(s.to_string()),
+        }
+    }
+}
+
+impl fmt::Display for DapAggregatorRole {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(match self {
+            Self::Helper => "helper",
+            Self::Leader => "leader",
+        })
+    }
+}
+
+impl From<DapAggregatorRole> for DapRole {
+    fn from(value: DapAggregatorRole) -> Self {
+        match value {
+            DapAggregatorRole::Leader => DapRole::Leader,
+            DapAggregatorRole::Helper => DapRole::Helper,
+        }
+    }
+}
+
 #[cfg(test)]
 mod test {
     use super::DapMediaType;
diff --git a/crates/daphne/src/hpke.rs b/crates/daphne/src/hpke.rs
index 193add9..a791781 100644
--- a/crates/daphne/src/hpke.rs
+++ b/crates/daphne/src/hpke.rs
@@ -17,6 +17,7 @@ use crate::{
     DapError, DapVersion,
 };
 use async_trait::async_trait;
+use info_and_aad::InfoAndAad;
 use serde::{Deserialize, Serialize};
 use std::{borrow::Borrow, ops::Deref};
 
@@ -176,13 +177,13 @@ impl HpkeConfig {
     /// configuration. The return values are the encapsulated key and the ciphertext.
     pub fn encrypt(
         &self,
-        info: &[u8],
-        aad: &[u8],
+        info: impl InfoAndAad,
         plaintext: &[u8],
     ) -> Result<HpkeCiphertext, DapError> {
         let mut sender: Hpke<ImplHpkeCrypto> = check_suite(self.kem_id, self.kdf_id, self.aead_id)?;
-        let (enc, mut ctx) = sender.setup_sender(&self.public_key, info, None, None, None)?;
-        let ciphertext = ctx.seal(aad, plaintext)?;
+        let (enc, mut ctx) =
+            sender.setup_sender(&self.public_key, &info.info_bytes(), None, None, None)?;
+        let ciphertext = ctx.seal(&info.aad_bytes().map_err(DapError::encoding)?, plaintext)?;
         Ok(HpkeCiphertext {
             config_id: self.id,
             enc,
@@ -193,17 +194,25 @@ impl HpkeConfig {
     pub(crate) fn decrypt(
         &self,
         private_key: &HpkePrivateKey,
-        info: &[u8],
-        aad: &[u8],
+        info: impl InfoAndAad,
         ciphertext: &HpkeCiphertext,
     ) -> Result<Vec<u8>, DapError> {
         if self.id != ciphertext.config_id {
             return Err(DapError::Transition(TransitionFailure::HpkeUnknownConfigId));
         }
         let receiver: Hpke<ImplHpkeCrypto> = check_suite(self.kem_id, self.kdf_id, self.aead_id)?;
-        let mut ctx =
-            receiver.setup_receiver(&ciphertext.enc, private_key, info, None, None, None)?;
-        let plaintext = ctx.open(aad, &ciphertext.payload)?;
+        let mut ctx = receiver.setup_receiver(
+            &ciphertext.enc,
+            private_key,
+            &info.info_bytes(),
+            None,
+            None,
+            None,
+        )?;
+        let plaintext = ctx.open(
+            &info.aad_bytes().map_err(DapError::encoding)?,
+            &ciphertext.payload,
+        )?;
         Ok(plaintext)
     }
 }
@@ -240,8 +249,7 @@ pub trait HpkeDecrypter {
     /// Decrypt the given HPKE ciphertext using the given info and AAD string.
     fn hpke_decrypt(
         &self,
-        info: &[u8],
-        aad: &[u8],
+        info: impl InfoAndAad,
         ciphertext: &HpkeCiphertext,
     ) -> Result<Vec<u8>, DapError>;
 }
@@ -252,11 +260,10 @@ where
 {
     fn hpke_decrypt(
         &self,
-        info: &[u8],
-        aad: &[u8],
+        info: impl InfoAndAad,
         ciphertext: &HpkeCiphertext,
     ) -> Result<Vec<u8>, DapError> {
-        <T as HpkeDecrypter>::hpke_decrypt(self, info, aad, ciphertext)
+        <T as HpkeDecrypter>::hpke_decrypt(self, info, ciphertext)
     }
 }
 
@@ -266,11 +273,10 @@ macro_rules! impl_decrypter_for_slice_types {
             impl$(<const $n: usize>)* HpkeDecrypter for $ty {
                 fn hpke_decrypt(
                     &self,
-                    info: &[u8],
-                    aad: &[u8],
+                    info: impl InfoAndAad,
                     ciphertext: &HpkeCiphertext,
                 ) -> Result<Vec<u8>, DapError> {
-                    self.iter().hpke_decrypt(info, aad, ciphertext)
+                    self.iter().hpke_decrypt(info, ciphertext)
                 }
             }
         )*
@@ -291,15 +297,14 @@ where
 {
     fn hpke_decrypt(
         &self,
-        info: &[u8],
-        aad: &[u8],
+        info: impl InfoAndAad,
         ciphertext: &HpkeCiphertext,
     ) -> Result<Vec<u8>, DapError> {
         self.clone()
             .map(|c| c.borrow())
             .find(|c| c.config.id == ciphertext.config_id)
             .ok_or(DapError::Transition(TransitionFailure::HpkeUnknownConfigId))?
-            .decrypt(info, aad, ciphertext)
+            .decrypt(info, ciphertext)
     }
 }
 
@@ -308,11 +313,10 @@ where
 impl HpkeDecrypter for HpkeReceiverConfig {
     fn hpke_decrypt(
         &self,
-        info: &[u8],
-        aad: &[u8],
+        info: impl InfoAndAad,
         ciphertext: &HpkeCiphertext,
     ) -> Result<Vec<u8>, DapError> {
-        [self].hpke_decrypt(info, aad, ciphertext)
+        [self].hpke_decrypt(info, ciphertext)
     }
 }
 
@@ -335,23 +339,20 @@ impl deepsize::DeepSizeOf for HpkeReceiverConfig {
 impl HpkeReceiverConfig {
     pub fn encrypt(
         &self,
-        info: &[u8],
-        aad: &[u8],
+        info: impl InfoAndAad,
         plaintext: &[u8],
     ) -> Result<HpkeCiphertext, DapError> {
-        self.config.encrypt(info, aad, plaintext)
+        self.config.encrypt(info, plaintext)
     }
 
     /// Decrypt `ciphertext` with info string `info` and associated data `aad` using this HPKE
     /// configuration and corresponding secret key. The return value is the plaintext.
     pub fn decrypt(
         &self,
-        info: &[u8],
-        aad: &[u8],
+        info: impl InfoAndAad,
         ciphertext: &HpkeCiphertext,
     ) -> Result<Vec<u8>, DapError> {
-        self.config
-            .decrypt(&self.private_key, info, aad, ciphertext)
+        self.config.decrypt(&self.private_key, info, ciphertext)
     }
 
     /// Generate and return a new HPKE receiver context given a HPKE config ID and HPKE KEM.
@@ -437,6 +438,135 @@ impl From<HpkePublicKeySerde> for HpkePublicKey {
     }
 }
 
+pub mod info_and_aad {
+    use crate::{
+        constants::{DapAggregatorRole, DapRole},
+        messages::{encode_u32_bytes, encode_u32_prefixed, BatchSelector, ReportMetadata, TaskId},
+        DapAggregationParam, DapVersion,
+    };
+    use prio::codec::{CodecError, Encode, ParameterizedEncode};
+
+    const CTX_INPUT_SHARE_DRAFT09: &[u8] = b"dap-09 input share";
+    const CTX_INPUT_SHARE_DRAFT13: &[u8] = b"dap-13 input share";
+    const CTX_AGG_SHARE_DRAFT09: &[u8] = b"dap-09 aggregate share";
+    const CTX_AGG_SHARE_DRAFT13: &[u8] = b"dap-13 aggregate share";
+
+    pub trait InfoAndAad {
+        // this could return a fixed size array, if GCE (generic constant expressions) was
+        // stabilized
+        //
+        // const LEN: usize;
+        //
+        // fn info_bytes(self) -> [u8; Self::LEN];
+
+        fn info_bytes(&self) -> Vec<u8>;
+
+        fn aad_bytes(&self) -> Result<Vec<u8>, CodecError>;
+    }
+
+    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+    pub struct InputShare<'s> {
+        // info
+        pub version: DapVersion,
+        pub receiver: DapAggregatorRole,
+        // aad
+        pub task_id: &'s TaskId,
+        pub report_metadata: &'s ReportMetadata,
+        pub public_share: &'s [u8],
+    }
+
+    // https://ietf-wg-ppm.github.io/draft-ietf-ppm-dap/draft-ietf-ppm-dap.html#section-4.5.2-14
+    impl InfoAndAad for InputShare<'_> {
+        fn info_bytes(&self) -> Vec<u8> {
+            into_bytes(
+                match self.version {
+                    DapVersion::Draft09 => CTX_INPUT_SHARE_DRAFT09,
+                    DapVersion::Latest => CTX_INPUT_SHARE_DRAFT13,
+                },
+                DapRole::Client,
+                self.receiver.into(),
+            )
+        }
+
+        fn aad_bytes(&self) -> Result<Vec<u8>, CodecError> {
+            let mut aad = Vec::with_capacity(
+                self.task_id.encoded_len().unwrap_or_default()
+                    + self
+                        .report_metadata
+                        .encoded_len_with_param(&self.version)
+                        .unwrap_or_default()
+                    + self.public_share.len(),
+            );
+            self.task_id.encode(&mut aad)?;
+            self.report_metadata
+                .encode_with_param(&self.version, &mut aad)?;
+            encode_u32_bytes(&mut aad, self.public_share)?;
+            Ok(aad)
+        }
+    }
+
+    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+    pub struct AggregateShare<'s> {
+        // info
+        pub version: DapVersion,
+        pub sender: DapAggregatorRole,
+        // aad
+        pub task_id: &'s TaskId,
+        pub agg_param: &'s DapAggregationParam,
+        pub batch_selector: &'s BatchSelector,
+    }
+
+    // https://ietf-wg-ppm.github.io/draft-ietf-ppm-dap/draft-ietf-ppm-dap.html#section-4.7.4-2
+    impl InfoAndAad for AggregateShare<'_> {
+        fn info_bytes(&self) -> Vec<u8> {
+            into_bytes(
+                match self.version {
+                    DapVersion::Draft09 => CTX_AGG_SHARE_DRAFT09,
+                    DapVersion::Latest => CTX_AGG_SHARE_DRAFT13,
+                },
+                self.sender.into(),
+                DapRole::Collector,
+            )
+        }
+
+        fn aad_bytes(&self) -> Result<Vec<u8>, CodecError> {
+            let mut aad = Vec::with_capacity(
+                self.task_id.encoded_len().unwrap_or_default()
+                    + self.agg_param.encoded_len().unwrap_or_default()
+                    + self
+                        .batch_selector
+                        .encoded_len_with_param(&self.version)
+                        .unwrap_or_default(),
+            );
+            self.task_id.encode(&mut aad)?;
+            encode_u32_prefixed(self.version, &mut aad, |_version, bytes| {
+                self.agg_param.encode(bytes)
+            })?;
+            self.batch_selector
+                .encode_with_param(&self.version, &mut aad)?;
+
+            Ok(aad)
+        }
+    }
+
+    fn into_bytes(constant: &'static [u8], sender: DapRole, receiver: DapRole) -> Vec<u8> {
+        fn role_to_id(role: DapRole) -> u8 {
+            match role {
+                DapRole::Collector => 0,
+                DapRole::Client => 1,
+                DapRole::Leader => 2,
+                DapRole::Helper => 3,
+            }
+        }
+
+        let len = constant.len() + 2;
+        let mut v = Vec::with_capacity(len);
+        v.extend_from_slice(constant);
+        v.extend([role_to_id(sender), role_to_id(receiver)]);
+        v
+    }
+}
+
 #[derive(Serialize, Deserialize)]
 #[serde(remote = "HpkePrivateKey")]
 struct HpkePrivateKeySerde(#[serde(getter = "HpkePrivateKeySerde::to_vec", with = "hex")] Vec<u8>);
@@ -455,33 +585,63 @@ impl From<HpkePrivateKeySerde> for HpkePrivateKey {
 
 #[cfg(test)]
 mod test {
-    use crate::hpke::{HpkeAeadId, HpkeConfig, HpkeKdfId, HpkeKemId, HpkeReceiverConfig};
+    use crate::{
+        constants::DapAggregatorRole,
+        hpke::{
+            info_and_aad::{self, InfoAndAad},
+            HpkeAeadId, HpkeConfig, HpkeKdfId, HpkeKemId, HpkeReceiverConfig,
+        },
+        messages::{
+            encode_u32_bytes, encode_u32_prefixed, BatchId, BatchSelector, Interval, ReportId,
+            ReportMetadata, TaskId,
+        },
+        test_versions, DapAggregationParam, DapVersion,
+    };
     use hpke_rs::{Hpke, HpkePrivateKey, HpkePublicKey, Mode};
     use hpke_rs_crypto::types::{AeadAlgorithm, KdfAlgorithm, KemAlgorithm};
     use hpke_rs_rust_crypto::HpkeRustCrypto as ImplHpkeCrypto;
-
-    #[test]
-    fn encrypt_roundtrip_x25519_hkdf_sha256() {
-        let info = b"info string";
-        let aad = b"associated data";
+    use prio::codec::{Encode as _, ParameterizedEncode as _};
+    use rand::seq::IteratorRandom;
+
+    fn encrypt_roundtrip_x25519_hkdf_sha256(version: DapVersion) {
+        let info = info_and_aad::InputShare {
+            version,
+            receiver: DapAggregatorRole::Helper,
+            task_id: &crate::messages::TaskId(rand::random()),
+            public_share: &[],
+            report_metadata: &ReportMetadata {
+                id: ReportId(rand::random()),
+                time: rand::random(),
+            },
+        };
         let plaintext = b"plaintext";
         let config = HpkeReceiverConfig::gen(23, HpkeKemId::X25519HkdfSha256).unwrap();
         println!("{}", serde_json::to_string(&config).unwrap());
-        let ciphertext = config.encrypt(info, aad, plaintext).unwrap();
-        assert_eq!(config.decrypt(info, aad, &ciphertext).unwrap(), plaintext);
+        let ciphertext = config.encrypt(info, plaintext).unwrap();
+        assert_eq!(config.decrypt(info, &ciphertext).unwrap(), plaintext);
     }
 
-    #[test]
-    fn encrypt_roundtrip_p256_hkdf_sha256() {
-        let info = b"info string";
-        let aad = b"associated data";
+    test_versions! { encrypt_roundtrip_x25519_hkdf_sha256 }
+
+    fn encrypt_roundtrip_p256_hkdf_sha256(version: DapVersion) {
+        let info = info_and_aad::AggregateShare {
+            version,
+            sender: DapAggregatorRole::Leader,
+            task_id: &crate::messages::TaskId(rand::random()),
+            agg_param: &crate::DapAggregationParam::Empty,
+            batch_selector: &crate::messages::BatchSelector::FixedSizeByBatchId {
+                batch_id: BatchId(rand::random()),
+            },
+        };
         let plaintext = b"plaintext";
         let config = HpkeReceiverConfig::gen(23, HpkeKemId::P256HkdfSha256).unwrap();
         println!("{}", serde_json::to_string(&config).unwrap());
-        let ciphertext = config.encrypt(info, aad, plaintext).unwrap();
-        assert_eq!(config.decrypt(info, aad, &ciphertext).unwrap(), plaintext);
+        let ciphertext = config.encrypt(info, plaintext).unwrap();
+        assert_eq!(config.decrypt(info, &ciphertext).unwrap(), plaintext);
     }
 
+    test_versions! { encrypt_roundtrip_p256_hkdf_sha256 }
+
     #[test]
     fn hpke_receiver_config_try_from() {
         let (private_key, public_key) = Hpke::<ImplHpkeCrypto>::new(
@@ -512,4 +672,124 @@ mod test {
         let bad_private_key = HpkePrivateKey::from(vec![0; 20]);
         assert!(HpkeReceiverConfig::try_from((config, bad_private_key)).is_err());
     }
+
+    // This code compares the old way to serializing info and aad parameters with the new
+    // abstracted way to ensure they don't differ.
+    fn manual_info_and_aad_equals_abstracted_one(version: DapVersion) {
+        const CTX_INPUT_SHARE_DRAFT09: &str = "dap-09 input share";
+        const CTX_AGG_SHARE_DRAFT09: &str = "dap-09 aggregate share";
+        // these were actually missing
+        const CTX_INPUT_SHARE_DRAFT13: &str = "dap-13 input share";
+        const CTX_AGG_SHARE_DRAFT13: &str = "dap-13 aggregate share";
+
+        const CTX_ROLE_COLLECTOR: u8 = 0;
+        const CTX_ROLE_CLIENT: u8 = 1;
+        const CTX_ROLE_LEADER: u8 = 2;
+        const CTX_ROLE_HELPER: u8 = 3;
+
+        let task_id = &TaskId(rand::random());
+        let agg_param = &DapAggregationParam::Empty;
+        let batch_selectors = &[
+            BatchSelector::TimeInterval {
+                batch_interval: Interval {
+                    start: rand::random(),
+                    duration: rand::random(),
+                },
+            },
+            BatchSelector::FixedSizeByBatchId {
+                batch_id: BatchId(rand::random()),
+            },
+        ];
+        let report_metadata = &ReportMetadata {
+            id: ReportId(rand::random()),
+            time: rand::random(),
+        };
+        let public_share = &vec![rand::random(); (0..100).choose(&mut rand::thread_rng()).unwrap()];
+
+        let input_share = |role, prefix: &str| {
+            let mut info = Vec::with_capacity(prefix.len() + 2);
+            info.extend_from_slice(prefix.as_bytes());
+            info.push(CTX_ROLE_CLIENT); // client
+            info.push(match role {
+                DapAggregatorRole::Leader => CTX_ROLE_LEADER,
+                DapAggregatorRole::Helper => CTX_ROLE_HELPER,
+            }); // Receiver role
+
+            let mut aad = Vec::with_capacity(58);
+            task_id.encode(&mut aad).unwrap();
+
+            report_metadata
+                .encode_with_param(&version, &mut aad)
+                .unwrap();
+            encode_u32_bytes(&mut aad, public_share).unwrap();
+
+            (info, aad)
+        };
+
+        let aggregate_share = |role, prefix: &str, batch_selector: &BatchSelector| {
+            let mut info = Vec::with_capacity(prefix.len() + 2);
+            info.extend_from_slice(prefix.as_bytes());
+            info.push(match role {
+                DapAggregatorRole::Leader => CTX_ROLE_LEADER,
+                DapAggregatorRole::Helper => CTX_ROLE_HELPER,
+            });
+            info.push(CTX_ROLE_COLLECTOR); // collector
+
+            let mut aad = Vec::with_capacity(40);
+            task_id.encode(&mut aad).unwrap();
+            encode_u32_prefixed(version, &mut aad, |_version, bytes| agg_param.encode(bytes))
+                .unwrap();
+            batch_selector
+                .encode_with_param(&version, &mut aad)
+                .unwrap();
+
+            (info, aad)
+        };
+
+        let (is, ag) = match version {
+            DapVersion::Draft09 => (CTX_INPUT_SHARE_DRAFT09, CTX_AGG_SHARE_DRAFT09),
+            DapVersion::Latest => (CTX_INPUT_SHARE_DRAFT13, CTX_AGG_SHARE_DRAFT13),
+        };
+
+        for role in [DapAggregatorRole::Leader, DapAggregatorRole::Helper] {
+            let new = info_and_aad::InputShare {
+                version,
+                receiver: role,
+                task_id,
+                report_metadata,
+                public_share,
+            };
+            let (info, aad) = input_share(role, is);
+            assert_eq!(info, new.info_bytes(), "info for ({role}, {is:?}) differs");
+            assert_eq!(
+                aad,
+                new.aad_bytes().unwrap(),
+                "aad for ({role}, {is:?}) differs"
+            );
+
+            for b in batch_selectors {
+                let new = info_and_aad::AggregateShare {
+                    version,
+                    sender: role,
+                    task_id,
+                    batch_selector: b,
+                    agg_param,
+                };
+
+                let (info, aad) = aggregate_share(role, ag, b);
+                assert_eq!(
+                    info,
+                    new.info_bytes(),
+                    "info for ({role}, {ag:?}, {b}) differs"
+                );
+                assert_eq!(
+                    aad,
+                    new.aad_bytes().unwrap(),
+                    "aad for ({role}, {ag:?}, {b}) differs"
+                );
+            }
+        }
+    }
+
+    test_versions! { manual_info_and_aad_equals_abstracted_one }
 }
diff --git a/crates/daphne/src/lib.rs b/crates/daphne/src/lib.rs
index e081434..45b4d74 100644
--- a/crates/daphne/src/lib.rs
+++ b/crates/daphne/src/lib.rs
@@ -818,6 +818,14 @@ impl Encode for DapAggregationParam {
             Self::Mastic(agg_param) => agg_param.encode(bytes),
         }
     }
+
+    fn encoded_len(&self) -> Option<usize> {
+        match self {
+            Self::Empty => Some(0),
+            #[cfg(feature = "experimental")]
+            Self::Mastic(agg_param) => agg_param.encoded_len(),
+        }
+    }
 }
 
 impl ParameterizedDecode<VdafConfig> for DapAggregationParam {
@@ -963,15 +971,6 @@ impl DapAggregateShare {
     }
 }
 
-/// DAP sender role.
-#[derive(Clone, Copy, Debug)]
-pub enum DapSender {
-    Client,
-    Collector,
-    Helper,
-    Leader,
-}
-
 /// Status of a collect job.
 #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "snake_case")]
diff --git a/crates/daphne/src/messages/mod.rs b/crates/daphne/src/messages/mod.rs
index 1e3e3db..d776d02 100644
--- a/crates/daphne/src/messages/mod.rs
+++ b/crates/daphne/src/messages/mod.rs
@@ -71,6 +71,10 @@ macro_rules! id_struct {
                 bytes.extend_from_slice(&self.0);
                 Ok(())
             }
+
+            fn encoded_len(&self) -> Option<usize> {
+                Some(self.0.len())
+            }
         }
 
         impl Decode for $sname {
diff --git a/crates/daphne/src/protocol/aggregator.rs b/crates/daphne/src/protocol/aggregator.rs
index f5a9e23..e106bd0 100644
--- a/crates/daphne/src/protocol/aggregator.rs
+++ b/crates/daphne/src/protocol/aggregator.rs
@@ -4,14 +4,15 @@
 #[cfg(feature = "experimental")]
 use crate::vdaf::mastic::{mastic_prep_finish, mastic_prep_finish_from_shares, mastic_prep_init};
 use crate::{
+    constants::DapAggregatorRole,
     error::DapAbort,
     fatal_error,
-    hpke::{HpkeConfig, HpkeDecrypter},
+    hpke::{info_and_aad, HpkeConfig, HpkeDecrypter},
     messages::{
-        self, encode_u32_bytes, encode_u32_prefixed, AggregationJobInitReq, AggregationJobResp,
-        Base64Encode, BatchSelector, Extension, HpkeCiphertext, PartialBatchSelector,
-        PlaintextInputShare, PrepareInit, Report, ReportId, ReportMetadata, ReportShare, TaskId,
-        Transition, TransitionFailure, TransitionVar,
+        self, encode_u32_bytes, AggregationJobInitReq, AggregationJobResp, Base64Encode,
+        BatchSelector, Extension, HpkeCiphertext, PartialBatchSelector, PlaintextInputShare,
+        PrepareInit, Report, ReportId, ReportMetadata, ReportShare, TaskId, Transition,
+        TransitionFailure, TransitionVar,
     },
     metrics::{DaphneMetrics, ReportStatus},
     vdaf::{
@@ -32,10 +33,6 @@ use std::{
     ops::Range,
 };
 
-use super::{
-    CTX_AGG_SHARE_DRAFT09, CTX_INPUT_SHARE_DRAFT09, CTX_ROLE_CLIENT, CTX_ROLE_COLLECTOR,
-    CTX_ROLE_HELPER, CTX_ROLE_LEADER,
-};
 use rayon::iter::{IntoParallelIterator, ParallelIterator as _};
 
 // Ping-pong message framing as defined in draft-irtf-cfrg-vdaf-08, Section 5.8. We do not
@@ -94,7 +91,7 @@ impl InitializedReport {
     pub(crate) fn new(
         decrypter: &impl HpkeDecrypter,
         valid_report_range: Range<messages::Time>,
-        is_leader: bool,
+        role: DapAggregatorRole,
         task_id: &TaskId,
         task_config: &DapTaskConfig,
         report_share: ReportShare,
@@ -124,24 +121,16 @@ impl InitializedReport {
             extensions,
             payload: input_share,
         } = {
-            let input_share_text = match task_config.version {
-                DapVersion::Draft09 | DapVersion::Latest => CTX_INPUT_SHARE_DRAFT09,
+            let info = info_and_aad::InputShare {
+                version: task_config.version,
+                receiver: role,
+                task_id,
+                report_metadata: &report_share.report_metadata,
+                public_share: &report_share.public_share,
             };
-            let mut info = Vec::with_capacity(input_share_text.len() + 2);
-            info.extend_from_slice(input_share_text);
-            info.push(CTX_ROLE_CLIENT); // Sender role (receiver role set below)
-            info.push([CTX_ROLE_HELPER, CTX_ROLE_LEADER][usize::from(is_leader)]); // Receiver role
-
-            let mut aad = Vec::with_capacity(58);
-            task_id.encode(&mut aad).map_err(DapError::encoding)?;
-            report_share
-                .report_metadata
-                .encode_with_param(&task_config.version, &mut aad)
-                .map_err(DapError::encoding)?;
-            encode_u32_bytes(&mut aad, &report_share.public_share).map_err(DapError::encoding)?;
 
             let encoded_input_share =
-                match decrypter.hpke_decrypt(&info, &aad, &report_share.encrypted_input_share) {
+                match decrypter.hpke_decrypt(info, &report_share.encrypted_input_share) {
                     Ok(encoded_input_share) => encoded_input_share,
                     Err(DapError::Transition(failure)) => {
                         return Ok(Self::Rejected {
@@ -199,7 +188,10 @@ impl InitializedReport {
             }
         };
 
-        let agg_id = usize::from(!is_leader);
+        let agg_id = match role {
+            DapAggregatorRole::Leader => 0,
+            DapAggregatorRole::Helper => 1,
+        };
         let res = match &task_config.vdaf {
             VdafConfig::Prio3(ref prio3_config) => prio3_prep_init(
                 prio3_config,
@@ -354,7 +346,7 @@ impl DapTaskConfig {
             let initialized_report = InitializedReport::new(
                 &decrypter,
                 valid_report_time_range.clone(),
-                true,
+                DapAggregatorRole::Leader,
                 task_id,
                 self,
                 ReportShare {
@@ -489,7 +481,7 @@ impl DapTaskConfig {
                 InitializedReport::new(
                     decrypter,
                     valid_report_time_range.clone(),
-                    false,
+                    DapAggregatorRole::Helper,
                     task_id,
                     self,
                     prep_init.report_share,
@@ -718,7 +710,7 @@ impl DapTaskConfig {
         version: DapVersion,
     ) -> Result<HpkeCiphertext, DapError> {
         produce_encrypted_agg_share(
-            true,
+            DapAggregatorRole::Leader,
             hpke_config,
             task_id,
             batch_sel,
@@ -740,7 +732,7 @@ impl DapTaskConfig {
         version: DapVersion,
     ) -> Result<HpkeCiphertext, DapError> {
         produce_encrypted_agg_share(
-            false,
+            DapAggregatorRole::Helper,
             hpke_config,
             task_id,
             batch_sel,
@@ -752,7 +744,7 @@ impl DapTaskConfig {
 }
 
 fn produce_encrypted_agg_share(
-    is_leader: bool,
+    role: DapAggregatorRole,
     hpke_config: &HpkeConfig,
     task_id: &TaskId,
     batch_sel: &BatchSelector,
@@ -767,26 +759,15 @@ fn produce_encrypted_agg_share(
         .get_encoded()
         .map_err(DapError::encoding)?;
 
-    let agg_share_text = CTX_AGG_SHARE_DRAFT09;
-    let n: usize = agg_share_text.len();
-    let mut info = Vec::with_capacity(n + 2);
-    info.extend_from_slice(agg_share_text);
-    info.push(if is_leader {
-        CTX_ROLE_LEADER
-    } else {
-        CTX_ROLE_HELPER
-    }); // Sender role
-    info.push(CTX_ROLE_COLLECTOR); // Receiver role
-
-    let mut aad = Vec::with_capacity(40);
-    task_id.encode(&mut aad).map_err(DapError::encoding)?;
-    encode_u32_prefixed(version, &mut aad, |_version, bytes| agg_param.encode(bytes))
-        .map_err(DapError::encoding)?;
-    batch_sel
-        .encode_with_param(&version, &mut aad)
-        .map_err(DapError::encoding)?;
+    let info = info_and_aad::AggregateShare {
+        version,
+        sender: role,
+        task_id,
+        agg_param,
+        batch_selector: batch_sel,
+    };
 
-    hpke_config.encrypt(&info, &aad, &agg_share_data)
+    hpke_config.encrypt(info, &agg_share_data)
 }
 
 /// checks if an iterator has no duplicate items, returns the ok if there are no dups or an error
diff --git a/crates/daphne/src/protocol/client.rs b/crates/daphne/src/protocol/client.rs
index 1a44e58..7559f22 100644
--- a/crates/daphne/src/protocol/client.rs
+++ b/crates/daphne/src/protocol/client.rs
@@ -4,19 +4,15 @@
 #[cfg(feature = "experimental")]
 use crate::vdaf::mastic::mastic_shard;
 use crate::{
-    hpke::HpkeConfig,
-    messages::{
-        encode_u32_bytes, Extension, PlaintextInputShare, Report, ReportId, ReportMetadata, TaskId,
-        Time,
-    },
+    constants::DapAggregatorRole,
+    hpke::{info_and_aad, HpkeConfig},
+    messages::{Extension, PlaintextInputShare, Report, ReportId, ReportMetadata, TaskId, Time},
     vdaf::{prio2::prio2_shard, prio3::prio3_shard, VdafError},
     DapError, DapMeasurement, DapVersion, VdafConfig,
 };
-use prio::codec::{Encode, ParameterizedEncode};
+use prio::codec::ParameterizedEncode;
 use rand::prelude::*;
 
-use super::{CTX_INPUT_SHARE_DRAFT09, CTX_ROLE_CLIENT, CTX_ROLE_HELPER, CTX_ROLE_LEADER};
-
 impl VdafConfig {
     /// Generate a report for a measurement. This method is run by the Client.
     ///
@@ -90,34 +86,25 @@ impl VdafConfig {
             plaintext_input_share.get_encoded_with_param(&version)
         });
 
-        let input_share_text = CTX_INPUT_SHARE_DRAFT09;
-        let n: usize = input_share_text.len();
-        let mut info = Vec::with_capacity(n + 2);
-        info.extend_from_slice(input_share_text);
-        info.push(CTX_ROLE_CLIENT); // Sender role
-        info.push(CTX_ROLE_LEADER); // Receiver role placeholder; updated below.
-
-        let mut aad = Vec::with_capacity(58);
-        task_id.encode(&mut aad).map_err(DapError::encoding)?;
-        metadata
-            .encode_with_param(&version, &mut aad)
-            .map_err(DapError::encoding)?;
-        encode_u32_bytes(&mut aad, &public_share).map_err(DapError::encoding)?;
+        let mut info = info_and_aad::InputShare {
+            version,
+            receiver: DapAggregatorRole::Leader, //placeholder; updated below.
+            task_id,
+            report_metadata: &metadata,
+            public_share: &public_share,
+        };
 
         let mut encrypted_input_shares = Vec::with_capacity(2);
         for (i, (hpke_config, encoded_input_share)) in
             hpke_configs.iter().zip(encoded_input_shares).enumerate()
         {
-            info[n + 1] = if i == 0 {
-                CTX_ROLE_LEADER
+            info.receiver = if i == 0 {
+                DapAggregatorRole::Leader
             } else {
-                CTX_ROLE_HELPER
+                DapAggregatorRole::Helper
             }; // Receiver role
-            let ciphertext = hpke_config.encrypt(
-                &info,
-                &aad,
-                &encoded_input_share.map_err(DapError::encoding)?,
-            )?;
+            let ciphertext =
+                hpke_config.encrypt(info, &encoded_input_share.map_err(DapError::encoding)?)?;
 
             encrypted_input_shares.push(ciphertext);
         }
diff --git a/crates/daphne/src/protocol/collector.rs b/crates/daphne/src/protocol/collector.rs
index 0bd5aa8..c143b09 100644
--- a/crates/daphne/src/protocol/collector.rs
+++ b/crates/daphne/src/protocol/collector.rs
@@ -4,15 +4,13 @@
 #[cfg(feature = "experimental")]
 use crate::vdaf::mastic::mastic_unshard;
 use crate::{
+    constants::DapAggregatorRole,
     fatal_error,
-    hpke::HpkeDecrypter,
-    messages::{encode_u32_prefixed, BatchSelector, HpkeCiphertext, TaskId},
+    hpke::{info_and_aad, HpkeDecrypter},
+    messages::{BatchSelector, HpkeCiphertext, TaskId},
     vdaf::{prio2::prio2_unshard, prio3::prio3_unshard},
     DapAggregateResult, DapAggregationParam, DapError, DapVersion, VdafConfig,
 };
-use prio::codec::{Encode, ParameterizedEncode};
-
-use super::{CTX_AGG_SHARE_DRAFT09, CTX_ROLE_COLLECTOR, CTX_ROLE_HELPER, CTX_ROLE_LEADER};
 
 impl VdafConfig {
     /// Decrypt and unshard a sequence of aggregate shares. This method is run by the Collector
@@ -47,30 +45,23 @@ impl VdafConfig {
             ));
         }
 
-        let agg_share_text = CTX_AGG_SHARE_DRAFT09;
-        let n: usize = agg_share_text.len();
-        let mut info = Vec::with_capacity(n + 2);
-        info.extend_from_slice(agg_share_text);
-        info.push(CTX_ROLE_LEADER); // Sender role placeholder
-        info.push(CTX_ROLE_COLLECTOR); // Receiver role
-
-        let mut aad = Vec::with_capacity(40);
-        task_id.encode(&mut aad).map_err(DapError::encoding)?;
-        encode_u32_prefixed(version, &mut aad, |_version, bytes| agg_param.encode(bytes))
-            .map_err(DapError::encoding)?;
-        batch_sel
-            .encode_with_param(&version, &mut aad)
-            .map_err(DapError::encoding)?;
+        let mut info = info_and_aad::AggregateShare {
+            version,
+            sender: DapAggregatorRole::Leader, // placeholder
+            task_id,
+            agg_param,
+            batch_selector: batch_sel,
+        };
 
         let mut agg_shares = Vec::with_capacity(encrypted_agg_shares.len());
         for (i, agg_share_ciphertext) in encrypted_agg_shares.iter().enumerate() {
-            info[n] = if i == 0 {
-                CTX_ROLE_LEADER
+            info.sender = if i == 0 {
+                DapAggregatorRole::Leader
             } else {
-                CTX_ROLE_HELPER
+                DapAggregatorRole::Helper
             };
 
-            let agg_share_data = decrypter.hpke_decrypt(&info, &aad, agg_share_ciphertext)?;
+            let agg_share_data = decrypter.hpke_decrypt(info, agg_share_ciphertext)?;
             agg_shares.push(agg_share_data);
         }
 
diff --git a/crates/daphne/src/protocol/mod.rs b/crates/daphne/src/protocol/mod.rs
index 46e08a1..7bfff85 100644
--- a/crates/daphne/src/protocol/mod.rs
+++ b/crates/daphne/src/protocol/mod.rs
@@ -5,17 +5,11 @@ pub(crate) mod aggregator;
 mod client;
 mod collector;
 
-const CTX_INPUT_SHARE_DRAFT09: &[u8] = b"dap-09 input share";
-const CTX_AGG_SHARE_DRAFT09: &[u8] = b"dap-09 aggregate share";
-const CTX_ROLE_COLLECTOR: u8 = 0;
-const CTX_ROLE_CLIENT: u8 = 1;
-const CTX_ROLE_LEADER: u8 = 2;
-const CTX_ROLE_HELPER: u8 = 3;
-
 #[cfg(test)]
 mod test {
     use crate::{
         assert_metrics_include,
+        constants::DapAggregatorRole,
         error::DapAbort,
         hpke::{HpkeAeadId, HpkeConfig, HpkeKdfId, HpkeKemId},
         messages::{
@@ -67,7 +61,7 @@ mod test {
         } = InitializedReport::new(
             &t.leader_hpke_receiver_config,
             t.valid_report_time_range(),
-            true, // is_leader
+            DapAggregatorRole::Leader,
             &t.task_id,
             &t.task_config,
             ReportShare {
@@ -90,7 +84,7 @@ mod test {
         } = InitializedReport::new(
             &t.helper_hpke_receiver_config,
             t.valid_report_time_range(),
-            false, // is_helper
+            DapAggregatorRole::Helper,
             &t.task_id,
             &t.task_config,
             ReportShare {
@@ -691,7 +685,7 @@ mod test {
         let initialized_report = InitializedReport::new(
             &t.leader_hpke_receiver_config,
             t.valid_report_time_range(),
-            true,
+            DapAggregatorRole::Leader,
             &t.task_id,
             &t.task_config,
             ReportShare {
@@ -741,7 +735,7 @@ mod test {
         let initialized_report = InitializedReport::new(
             &t.leader_hpke_receiver_config,
             t.valid_report_time_range(),
-            true,
+            DapAggregatorRole::Leader,
             &t.task_id,
             &t.task_config,
             ReportShare {


Test patch:
diff --git a/crates/daphne-server/src/router/test_routes.rs b/crates/daphne-server/src/router/test_routes.rs
index bf9120b..b09c42f 100644
--- a/crates/daphne-server/src/router/test_routes.rs
+++ b/crates/daphne-server/src/router/test_routes.rs
@@ -11,23 +11,21 @@ use axum::{
     Json,
 };
 use daphne::{
+    constants::DapAggregatorRole,
     hpke::HpkeReceiverConfig,
     messages::{Base64Encode, TaskId},
     roles::{leader, DapLeader},
     DapVersion,
 };
-use daphne_service_utils::{
-    test_route_types::{InternalTestAddTask, InternalTestEndpointForTask},
-    DapRole,
-};
+use daphne_service_utils::test_route_types::{InternalTestAddTask, InternalTestEndpointForTask};
 use serde::Deserialize;
 
 use crate::App;
 
 use super::{AxumDapResponse, DaphneService};
 
-pub fn add_test_routes(router: super::Router<App>, role: DapRole) -> super::Router<App> {
-    let router = if role == DapRole::Leader {
+pub fn add_test_routes(router: super::Router<App>, role: DapAggregatorRole) -> super::Router<App> {
+    let router = if role == DapAggregatorRole::Leader {
         router
             .route("/internal/process", post(leader_process))
             .route(
diff --git a/crates/daphne-server/tests/e2e/test_runner.rs b/crates/daphne-server/tests/e2e/test_runner.rs
index aeb6bad..b164d94 100644
--- a/crates/daphne-server/tests/e2e/test_runner.rs
+++ b/crates/daphne-server/tests/e2e/test_runner.rs
@@ -4,7 +4,7 @@
 use anyhow::Context;
 use assert_matches::assert_matches;
 use daphne::{
-    constants::DapMediaType,
+    constants::{DapAggregatorRole, DapMediaType},
     hpke::{HpkeAeadId, HpkeConfig, HpkeKdfId, HpkeKemId, HpkeReceiverConfig},
     messages::{
         encode_base64url, taskprov::TaskprovAdvertisement, Base64Encode, BatchId, CollectionJobId,
@@ -556,15 +556,14 @@ impl TestRunner {
 
     async fn post_internal<I: Serialize, O: for<'a> Deserialize<'a> + Any>(
         &self,
-        is_leader: bool,
+        role: DapAggregatorRole,
         path: &str,
         data: &I,
     ) -> anyhow::Result<O> {
         let client = self.http_client();
-        let mut url = if is_leader {
-            self.leader_url.clone()
-        } else {
-            self.helper_url.clone()
+        let mut url = match role {
+            DapAggregatorRole::Leader => self.leader_url.clone(),
+            DapAggregatorRole::Helper => self.helper_url.clone(),
         };
         url.set_path(path); // Overwrites the version path (i.e., "/v09")
         let resp = client
@@ -602,7 +601,8 @@ impl TestRunner {
         path: &str,
         data: &I,
     ) -> anyhow::Result<O> {
-        self.post_internal(true /* is_leader */, path, data).await
+        self.post_internal(DapAggregatorRole::Leader, path, data)
+            .await
     }
 
     pub async fn helper_post_internal<I: Serialize, O: for<'a> Deserialize<'a> + Any>(
@@ -610,7 +610,8 @@ impl TestRunner {
         path: &str,
         data: &I,
     ) -> anyhow::Result<O> {
-        self.post_internal(false /* is_leader */, path, data).await
+        self.post_internal(DapAggregatorRole::Helper, path, data)
+            .await
     }
 
     pub async fn internal_delete_all(&self, batch_interval: &Interval) -> anyhow::Result<()> {
diff --git a/crates/daphne-service-utils/src/test_route_types.rs b/crates/daphne-service-utils/src/test_route_types.rs
index 904a929..7bba506 100644
--- a/crates/daphne-service-utils/src/test_route_types.rs
+++ b/crates/daphne-service-utils/src/test_route_types.rs
@@ -6,6 +6,7 @@
 //! [interop]: https://divergentdave.github.io/draft-dcook-ppm-dap-interop-test-design/draft-dcook-ppm-dap-interop-test-design.html
 
 use daphne::{
+    constants::DapAggregatorRole,
     messages::{Duration, TaskId, Time},
     vdaf::{Prio3Config, VdafConfig},
 };
@@ -16,7 +17,7 @@ use url::Url;
 #[derive(Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub struct InternalTestEndpointForTask {
-    pub role: super::DapRole,
+    pub role: DapAggregatorRole,
 }
 
 #[derive(Serialize, Deserialize)]
@@ -82,7 +83,7 @@ pub struct InternalTestAddTask {
     pub leader_authentication_token: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub collector_authentication_token: Option<String>,
-    pub role: super::DapRole,
+    pub role: DapAggregatorRole,
     pub vdaf_verify_key: String, // base64url
     pub query_type: u8,
     pub min_batch_size: u64,