aggregator API: default to bearer tokens

Previously, the aggregator API assumed that any aggregator or collector
auth tokens it was handling were `AuthenticationToken::DapAuth`. Janus
must support those for the interop testing API and to work with Daphne,
but generally, we prefer to use more boring `Authorization: Bearer`
tokens. Certainly any tasks provisioned via the aggregator API and
`divviup-api` should use bearer tokens.

With this PR, we augment the `PostTaskReq` and `PostTaskResp` messages
so that they use `AuthenticationToken` to represent aggregator and
collector tokens, meaning that on the wire, a token now looks like:

```
{
  "type": "Bearer",
  "token": "AAAAAAAAA<etc.>"
}
```

Also, when the aggregator API generates either kind of token, it now
generates a bearer token. I'm not aware of any scenarios where we need
to generate `DapAuth` tokens in the aggregator API so there is no
affordance for requesting that kind of token.

See #472

Cargo.toml

@@ -45,6 +45,8 @@ opentelemetry = { version = "0.19", features = ["metrics"] }
 prio = { version = "0.12.2", features = ["multithreaded"] }
 serde = { version = "1.0.175", features = ["derive"] }
 rstest = "0.17.0"
+thiserror = "1.0"
+tokio = { version = "1.29", features = ["full", "tracing"] }
 trillium = "0.2.9"
 trillium-api = { version = "0.2.0-rc.3", default-features = false }
 trillium-caching-headers = "0.2.1"

aggregator/Cargo.toml

@@ -82,7 +82,7 @@ signal-hook = "0.3.17"
 signal-hook-tokio = { version = "0.3.1", features = ["futures-v0_3"] }
 testcontainers = { version = "0.14.0", optional = true }
 thiserror = "1.0"
-tokio = { version = "1.29", features = ["full", "tracing"] }
+tokio.workspace = true
 tokio-postgres = { version = "0.7.8", features = ["with-chrono-0_4", "with-serde_json-1", "with-uuid-1", "array-impls"] }
 tonic = { version = "0.8", optional = true, features = ["tls", "tls-webpki-roots"] }                                      # keep this version in sync with what opentelemetry-otlp uses
 tracing = "0.1.37"

aggregator/src/aggregator/http_handlers.rs

@@ -4,7 +4,7 @@ use async_trait::async_trait;
 use janus_aggregator_core::{datastore::Datastore, instrumented};
 use janus_core::{
     http::extract_bearer_token,
-    task::{AuthenticationToken, DapAuthToken, DAP_AUTH_HEADER},
+    task::{AuthenticationToken, DAP_AUTH_HEADER},
     time::Clock,
 };
 use janus_messages::{
@@ -538,14 +538,13 @@ fn parse_auth_token(task_id: &TaskId, conn: &Conn) -> Result<Option<Authenticati
     if let Some(bearer_token) =
         extract_bearer_token(conn).map_err(|_| Error::UnauthorizedRequest(*task_id))?
     {
-        return Ok(Some(AuthenticationToken::Bearer(bearer_token)));
+        return Ok(Some(bearer_token));
     }
 
     conn.request_headers()
         .get(DAP_AUTH_HEADER)
         .map(|value| {
-            DapAuthToken::try_from(value.as_ref().to_vec())
-                .map(AuthenticationToken::DapAuth)
+            AuthenticationToken::new_dap_auth_token_from_bytes(value.as_ref())
                 .map_err(|e| Error::BadRequest(format!("bad DAP-Auth-Token header: {e}")))
         })
         .transpose()
@@ -1376,7 +1375,7 @@ mod tests {
             .aggregator_auth_tokens()
             .iter()
             .find(|token| matches!(token, AuthenticationToken::DapAuth(_)))
-            .map(|token| AuthenticationToken::Bearer(token.as_ref().to_vec()))
+            .map(|token| AuthenticationToken::new_bearer_token_from_bytes(token.as_ref()).unwrap())
             .unwrap();
 
         for auth_token in [Some(wrong_token_value), Some(wrong_token_format), None] {

aggregator/src/bin/aggregator.rs

@@ -1,5 +1,4 @@
 use anyhow::{Context, Result};
-use base64::{engine::general_purpose::STANDARD, Engine};
 use clap::Parser;
 use janus_aggregator::{
     aggregator::{self, garbage_collector::GarbageCollector, http_handlers::aggregator_handler},
@@ -11,8 +10,8 @@ use janus_aggregator::{
     config::{BinaryConfig, CommonConfig, TaskprovConfig},
 };
 use janus_aggregator_api::{self, aggregator_api_handler};
-use janus_aggregator_core::{datastore::Datastore, SecretBytes};
-use janus_core::time::RealClock;
+use janus_aggregator_core::datastore::Datastore;
+use janus_core::{task::AuthenticationToken, time::RealClock};
 use serde::{Deserialize, Serialize};
 use std::{
     future::{ready, Future},
@@ -154,11 +153,9 @@ fn build_aggregator_api_handler<'a>(
         .iter()
         .filter(|token| !token.is_empty())
         .map(|token| {
-            let token_bytes = STANDARD
-                .decode(token)
-                .context("couldn't base64-decode aggregator API auth token")?;
-
-            Ok(SecretBytes::new(token_bytes))
+            // Aggregator API auth tokens are always bearer tokens
+            AuthenticationToken::new_bearer_token_from_string(token)
+                .context("invalid aggregator API auth token")
         })
         .collect::<Result<Vec<_>>>()?;