Honor DAP-Auth-Token or Authorization: Bearer

Cargo.toml

@@ -23,6 +23,7 @@ rust-version = "1.65.0"
 version = "0.4.6"
 
 [workspace.dependencies]
+anyhow = "1"
 # Disable default features to disable compatibility with the old `time` crate, and we also don't
 # (yet) need other default features.
 # https://docs.rs/chrono/latest/chrono/#duration
@@ -41,6 +42,7 @@ janus_messages = { version = "0.4", path = "messages" }
 k8s-openapi = { version = "0.18.0", features = ["v1_24"] }  # keep this version in sync with what is referenced by the indirect dependency via `kube`
 kube = { version = "0.82.2", default-features = false, features = ["client", "rustls-tls"] }
 prio = { version = "0.12.1", features = ["multithreaded"] }
+rstest = "0.17.0"
 trillium = "0.2.9"
 trillium-api = { version = "0.2.0-rc.3", default-features = false }
 trillium-caching-headers = "0.2.1"

aggregator/Cargo.toml

@@ -35,7 +35,7 @@ test-util = [
 
 [dependencies]
 async-trait = "0.1"
-anyhow = "1"
+anyhow.workspace = true
 atty = "0.2"
 backoff = { version = "0.4.0", features = ["tokio"] }
 base64 = "0.21.0"
@@ -110,6 +110,7 @@ hyper = "0.14.26"
 janus_aggregator = { path = ".", features = ["fpvec_bounded_l2", "test-util"] }
 janus_aggregator_core = { workspace = true, features = ["test-util"] }
 mockito = "1.0.2"
+rstest.workspace = true
 tempfile = "3.5.0"
 tokio = { version = "1", features = ["test-util"] } # ensure this remains compatible with the non-dev dependency
 trillium-testing.workspace = true

aggregator/src/aggregator.rs

@@ -2358,6 +2358,10 @@ async fn send_request_to_helper<T: Encode>(
     let response_result = http_client
         .request(method, url)
         .header(CONTENT_TYPE, content_type)
+        // TODO(#472): We want to be able to communicate with new Janus (prefers bearer token but
+        // supports `DAP-Auth-Token`) as well as older Janus and Daphne (which require
+        // `DAP-Auth-Token`) so for the moment, we send `DAP-Auth-Token`. But eventually we should
+        // determine the appropriate token header to send for a given task.
         .header(DAP_AUTH_HEADER, auth_token.as_ref())
         .body(request_body)
         .send()

aggregator/src/aggregator/aggregate_init_tests.rs

@@ -9,7 +9,7 @@ use janus_aggregator_core::{
     task::{test_util::TaskBuilder, QueryType, Task},
 };
 use janus_core::{
-    task::VdafInstance,
+    task::{VdafInstance, DAP_AUTH_HEADER},
     test_util::{dummy_vdaf, install_test_trace_subscriber, run_vdaf, VdafTranscript},
     time::{Clock, MockClock, TimeExt as _},
 };
@@ -89,13 +89,29 @@ pub(super) struct AggregationJobInitTestCase {
     pub(super) report_share_generator: ReportShareGenerator,
     pub(super) report_shares: Vec<ReportShare>,
     pub(super) aggregation_job_id: AggregationJobId,
+    aggregation_job_init_req: AggregationJobInitializeReq<TimeInterval>,
     pub(super) aggregation_param: dummy_vdaf::AggregationParam,
     pub(super) handler: Box<dyn Handler>,
     pub(super) datastore: Arc<Datastore<MockClock>>,
     _ephemeral_datastore: EphemeralDatastore,
 }
 
 pub(super) async fn setup_aggregate_init_test() -> AggregationJobInitTestCase {
+    let test_case = setup_aggregate_init_test_without_sending_request().await;
+
+    let response = put_aggregation_job(
+        &test_case.task,
+        &test_case.aggregation_job_id,
+        &test_case.aggregation_job_init_req,
+        &test_case.handler,
+    )
+    .await;
+    assert_eq!(response.status(), Some(Status::Ok));
+
+    test_case
+}
+
+async fn setup_aggregate_init_test_without_sending_request() -> AggregationJobInitTestCase {
     install_test_trace_subscriber();
 
     let task = TaskBuilder::new(QueryType::TimeInterval, VdafInstance::Fake, Role::Helper).build();
@@ -125,21 +141,13 @@ pub(super) async fn setup_aggregate_init_test() -> AggregationJobInitTestCase {
         report_shares.clone(),
     );
 
-    let response = put_aggregation_job(
-        &task,
-        &aggregation_job_id,
-        &aggregation_job_init_req,
-        &handler,
-    )
-    .await;
-    assert_eq!(response.status(), Some(Status::Ok));
-
     AggregationJobInitTestCase {
         clock,
         task,
         report_shares,
         report_share_generator,
         aggregation_job_id,
+        aggregation_job_init_req,
         aggregation_param,
         handler: Box::new(handler),
         datastore,
@@ -155,7 +163,7 @@ pub(crate) async fn put_aggregation_job(
 ) -> TestConn {
     put(task.aggregation_job_uri(aggregation_job_id).unwrap().path())
         .with_request_header(
-            "DAP-Auth-Token",
+            DAP_AUTH_HEADER,
             task.primary_aggregator_auth_token().as_ref().to_owned(),
         )
         .with_request_header(
@@ -167,6 +175,69 @@ pub(crate) async fn put_aggregation_job(
         .await
 }
 
+#[tokio::test]
+async fn aggregation_job_init_authorization_bearer_header() {
+    let test_case = setup_aggregate_init_test_without_sending_request().await;
+
+    let response = put(test_case
+        .task
+        .aggregation_job_uri(&test_case.aggregation_job_id)
+        .unwrap()
+        .path())
+    // Authenticate using an "Authorization: Bearer <token>" header instead of "DAP-Auth-Token"
+    .with_request_header(
+        KnownHeaderName::Authorization,
+        test_case
+            .task
+            .primary_aggregator_auth_token()
+            .bearer_token(),
+    )
+    .with_request_header(
+        KnownHeaderName::ContentType,
+        AggregationJobInitializeReq::<TimeInterval>::MEDIA_TYPE,
+    )
+    .with_request_body(test_case.aggregation_job_init_req.get_encoded())
+    .run_async(&test_case.handler)
+    .await;
+
+    assert_eq!(response.status(), Some(Status::Ok));
+}
+
+#[rstest::rstest]
+#[case::not_bearer_token("wrong kind of token")]
+#[case::not_base64("Bearer: ")]
+#[tokio::test]
+async fn aggregation_job_init_malformed_authorization_header(#[case] header_value: &'static str) {
+    let test_case = setup_aggregate_init_test_without_sending_request().await;
+
+    let response = put(test_case
+        .task
+        .aggregation_job_uri(&test_case.aggregation_job_id)
+        .unwrap()
+        .path())
+    // Authenticate using a malformed "Authorization: Bearer <token>" header and a `DAP-Auth-Token`
+    // header. The presence of the former should cause an error despite the latter being present and
+    // well formed.
+    .with_request_header(KnownHeaderName::Authorization, header_value.to_string())
+    .with_request_header(
+        DAP_AUTH_HEADER,
+        test_case
+            .task
+            .primary_aggregator_auth_token()
+            .as_ref()
+            .to_owned(),
+    )
+    .with_request_header(
+        KnownHeaderName::ContentType,
+        AggregationJobInitializeReq::<TimeInterval>::MEDIA_TYPE,
+    )
+    .with_request_body(test_case.aggregation_job_init_req.get_encoded())
+    .run_async(&test_case.handler)
+    .await;
+
+    assert_eq!(response.status(), Some(Status::BadRequest));
+}
+
 #[tokio::test]
 async fn aggregation_job_mutation_aggregation_job() {
     let test_case = setup_aggregate_init_test().await;

aggregator/src/aggregator/collection_job_driver.rs

@@ -15,8 +15,7 @@ use janus_aggregator_core::{
     },
     task,
 };
-use janus_core::time::Clock;
-use janus_core::vdaf_dispatch;
+use janus_core::{time::Clock, vdaf_dispatch};
 use janus_messages::{
     query_type::{FixedSize, QueryType, TimeInterval},
     AggregateShare, AggregateShareReq, BatchSelector,

aggregator/src/aggregator/http_handlers.rs

@@ -4,6 +4,7 @@ use async_trait::async_trait;
 use janus_aggregator_api::instrumented;
 use janus_aggregator_core::datastore::Datastore;
 use janus_core::{
+    http::extract_bearer_token,
     task::{AuthenticationToken, DAP_AUTH_HEADER},
     time::Clock,
 };
@@ -358,7 +359,7 @@ async fn aggregation_jobs_put<C: Clock>(
 
     let task_id = parse_task_id(&captures)?;
     let aggregation_job_id = parse_aggregation_job_id(&captures)?;
-    let auth_token = parse_auth_token(conn);
+    let auth_token = parse_auth_token(&task_id, conn)?;
     let response = aggregator
         .handle_aggregate_init(&task_id, &aggregation_job_id, &body, auth_token)
         .await?;
@@ -379,7 +380,7 @@ async fn aggregation_jobs_post<C: Clock>(
 
     let task_id = parse_task_id(&captures)?;
     let aggregation_job_id = parse_aggregation_job_id(&captures)?;
-    let auth_token = parse_auth_token(conn);
+    let auth_token = parse_auth_token(&task_id, conn)?;
     let response = aggregator
         .handle_aggregate_continue(&task_id, &aggregation_job_id, &body, auth_token)
         .await?;
@@ -400,7 +401,7 @@ async fn collection_jobs_put<C: Clock>(
 
     let task_id = parse_task_id(&captures)?;
     let collection_job_id = parse_collection_job_id(&captures)?;
-    let auth_token = parse_auth_token(conn);
+    let auth_token = parse_auth_token(&task_id, conn)?;
     aggregator
         .handle_create_collection_job(&task_id, &collection_job_id, &body, auth_token)
         .await?;
@@ -418,7 +419,7 @@ async fn collection_jobs_post<C: Clock>(
 ) -> Result<(), Error> {
     let task_id = parse_task_id(&captures)?;
     let collection_job_id = parse_collection_job_id(&captures)?;
-    let auth_token = parse_auth_token(conn);
+    let auth_token = parse_auth_token(&task_id, conn)?;
     let response_opt = aggregator
         .handle_get_collection_job(&task_id, &collection_job_id, auth_token)
         .await?;
@@ -446,7 +447,7 @@ async fn collection_jobs_delete<C: Clock>(
 ) -> Result<Status, Error> {
     let task_id = parse_task_id(&captures)?;
     let collection_job_id = parse_collection_job_id(&captures)?;
-    let auth_token = parse_auth_token(conn);
+    let auth_token = parse_auth_token(&task_id, conn)?;
     aggregator
         .handle_delete_collection_job(&task_id, &collection_job_id, auth_token)
         .await?;
@@ -465,7 +466,7 @@ async fn aggregate_shares<C: Clock>(
     validate_content_type(conn, AggregateShareReq::<TimeInterval>::MEDIA_TYPE)?;
 
     let task_id = parse_task_id(&captures)?;
-    let auth_token = parse_auth_token(conn);
+    let auth_token = parse_auth_token(&task_id, conn)?;
     let share = aggregator
         .handle_aggregate_share(&task_id, &body, auth_token)
         .await?;
@@ -521,11 +522,19 @@ fn parse_collection_job_id(captures: &Captures) -> Result<CollectionJobId, Error
         .map_err(|_| Error::BadRequest("invalid CollectionJobId".to_owned()))
 }
 
-/// Get the value of the DAP-Auth-Token header from the request.
-fn parse_auth_token(conn: &Conn) -> Option<AuthenticationToken> {
-    conn.request_headers()
+/// Get the authorization token header from the request.
+fn parse_auth_token(task_id: &TaskId, conn: &Conn) -> Result<Option<AuthenticationToken>, Error> {
+    // Prefer a bearer token, then fall back to DAP-Auth-Token
+    let bearer_token =
+        extract_bearer_token(conn).map_err(|_| Error::UnauthorizedRequest(*task_id))?;
+    if bearer_token.is_some() {
+        return Ok(bearer_token);
+    }
+
+    Ok(conn
+        .request_headers()
         .get(DAP_AUTH_HEADER)
-        .map(|value| value.as_ref().to_owned().into())
+        .map(|value| value.as_ref().to_owned().into()))
 }
 
 #[cfg(test)]

aggregator_api/Cargo.toml

@@ -9,7 +9,7 @@ rust-version.workspace = true
 version.workspace = true
 
 [dependencies]
-anyhow = "1"
+anyhow.workspace = true
 async-trait = "0.1"
 base64 = "0.21.0"
 janus_aggregator_core.workspace = true

aggregator_api/src/lib.rs

@@ -1,13 +1,14 @@
 //! This crate implements the Janus Aggregator API.
 
 use crate::models::{GetTaskIdsResp, PostTaskReq};
-use base64::{engine::general_purpose::STANDARD, Engine};
 use janus_aggregator_core::{
     datastore::{self, Datastore},
     task::Task,
     SecretBytes,
 };
-use janus_core::{hpke::generate_hpke_config_and_private_key, time::Clock};
+use janus_core::{
+    hpke::generate_hpke_config_and_private_key, http::extract_bearer_token, time::Clock,
+};
 use janus_messages::{Duration, HpkeAeadId, HpkeKdfId, HpkeKemId, Role, TaskId, Time};
 use models::{GetTaskMetricsResp, TaskResp};
 use querystring::querify;
@@ -52,21 +53,18 @@ pub fn aggregator_api_handler<C: Clock>(ds: Arc<Datastore<C>>, cfg: Config) -> i
 }
 
 async fn auth_check(conn: &mut Conn, State(cfg): State<Arc<Config>>) -> impl Handler {
-    if let Some(authorization_value) = conn.headers().get("authorization") {
-        if let Some(received_token) = authorization_value.as_ref().strip_prefix(b"Bearer ") {
-            let decoded = match STANDARD.decode(received_token) {
-                Ok(decoded) => decoded,
-                Err(_) => {
-                    return Some((Status::Unauthorized, Halt));
-                }
-            };
-            if cfg.auth_tokens.iter().any(|key| {
-                constant_time::verify_slices_are_equal(decoded.as_ref(), key.as_ref()).is_ok()
-            }) {
-                // Authorization succeeds.
-                return None;
-            }
+    let bearer_token = match extract_bearer_token(conn) {
+        Ok(Some(t)) => t,
+        _ => {
+            return Some((Status::Unauthorized, Halt));
         }
+    };
+
+    if cfg.auth_tokens.iter().any(|key| {
+        constant_time::verify_slices_are_equal(bearer_token.as_ref(), key.as_ref()).is_ok()
+    }) {
+        // Authorization succeeds.
+        return None;
     }
 
     // Authorization fails.

aggregator_core/Cargo.toml

@@ -20,8 +20,8 @@ test-util = [
 ]
 
 [dependencies]
+anyhow.workspace = true
 async-trait = "0.1"
-anyhow = "1"
 backoff = { version = "0.4.0", features = ["tokio"] }
 base64 = "0.21.0"
 bytes = "1.4.0"
@@ -66,7 +66,7 @@ async-std = { version = "1.12.0",  features = ["attributes"] }
 hyper = "0.14.26"
 janus_aggregator_core = { path = ".", features = ["test-util"] }
 janus_core = { workspace = true, features = ["test-util"] }
-rstest = "0.17.0"
+rstest.workspace = true
 rstest_reuse = "0.5.0"
 serde_test = "1.0.160"
 tempfile = "3.5.0"

core/Cargo.toml

@@ -30,6 +30,7 @@ test-util = [
 ]
 
 [dependencies]
+anyhow.workspace = true
 assert_matches = { version = "1", optional = true }
 backoff = { version = "0.4.0", features = ["tokio"] }
 base64 = "0.21.0"
@@ -57,6 +58,7 @@ tokio = { version = "1.27", features = ["macros", "net", "rt"] }
 tracing = "0.1.37"
 tracing-log = { version = "0.1.3", optional = true }
 tracing-subscriber = { version = "0.3", features = ["std", "env-filter", "fmt"], optional = true }
+trillium.workspace = true
 
 [dev-dependencies]
 fixed = "1.23"