encrypt bearer tokens at rest

Cargo.toml

@@ -17,6 +17,7 @@ integration-testing = []
 members = [".", "migration", "client", "test-support", "cli"]
 
 [dependencies]
+aes-gcm = "0.10.2"
 async-lock = "2.7.0"
 async-session = "3.0.0"
 base64 = "0.21.2"
@@ -60,6 +61,7 @@ trillium-sessions = "0.4.2"
 trillium-static-compiled = "0.5.0"
 trillium-testing = { version = "0.5.0", optional = true }
 trillium-tokio = "0.3.1"
+typenum = "1.16.0"
 url = "2.4.0"
 uuid = { version = "1.4.1", features = ["v4", "fast-rng", "serde"] }
 validator = { version = "0.16.1", features = ["derive"] }

README.md

@@ -38,6 +38,7 @@ An example `.envrc` is provided for optional but recommended use with [`direnv`]
 * `DATABASE_URL` -- A [libpq-compatible postgres uri](https://www.postgresql.org/docs/current/libpq-connect.html#id-1.7.3.8.3.6)
 * `POSTMARK_TOKEN` -- the token from the transactional stream from a [postmark](https://postmarkapp.com) account
 * `EMAIL_ADDRESS` -- the address this deployment should send from
+* `DATABASE_ENCRYPTION_KEYS` -- Comma-joined url-safe-no-pad base64'ed 16 byte cryptographically-random keys. The first one will be used to encrypt aggregator API authentication tokens at rest in the database
 
 ### Optional binding environment variables
 

client/tests/aggregators.rs

@@ -95,7 +95,8 @@ async fn rotate_bearer_token(
             .one(app.db())
             .await?
             .unwrap()
-            .bearer_token,
+            .bearer_token(app.crypter())
+            .unwrap(),
         new_bearer_token
     );
     Ok(())

migration/src/lib.rs

@@ -15,6 +15,7 @@ mod m20230626_183248_add_is_first_party_to_aggregators;
 mod m20230630_175314_create_api_tokens;
 mod m20230703_201332_add_additional_fields_to_api_tokens;
 mod m20230725_220134_add_vdafs_and_query_types_to_aggregators;
+mod m20230731_181722_rename_aggregator_bearer_token;
 
 pub struct Migrator;
 
@@ -37,6 +38,7 @@ impl MigratorTrait for Migrator {
             Box::new(m20230630_175314_create_api_tokens::Migration),
             Box::new(m20230703_201332_add_additional_fields_to_api_tokens::Migration),
             Box::new(m20230725_220134_add_vdafs_and_query_types_to_aggregators::Migration),
+            Box::new(m20230731_181722_rename_aggregator_bearer_token::Migration),
         ]
     }
 }

migration/src/m20230731_181722_rename_aggregator_bearer_token.rs

@@ -0,0 +1,64 @@
+use sea_orm_migration::prelude::*;
+
+#[derive(DeriveMigrationName)]
+pub struct Migration;
+
+#[async_trait::async_trait]
+impl MigrationTrait for Migration {
+    async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .exec_stmt(Table::truncate().table(Task::Table).to_owned())
+            .await?;
+
+        manager
+            .exec_stmt(Table::truncate().table(Aggregator::Table).to_owned())
+            .await?;
+
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(Aggregator::Table)
+                    .add_column(
+                        ColumnDef::new(Aggregator::EncryptedBearerToken)
+                            .binary()
+                            .not_null(),
+                    )
+                    .drop_column(Aggregator::BearerToken)
+                    .to_owned(),
+            )
+            .await?;
+        Ok(())
+    }
+
+    async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
+        manager
+            .exec_stmt(Table::truncate().table(Task::Table).to_owned())
+            .await?;
+        manager
+            .exec_stmt(Table::truncate().table(Aggregator::Table).to_owned())
+            .await?;
+
+        manager
+            .alter_table(
+                Table::alter()
+                    .table(Aggregator::Table)
+                    .add_column(ColumnDef::new(Aggregator::BearerToken).string().not_null())
+                    .drop_column(Aggregator::EncryptedBearerToken)
+                    .to_owned(),
+            )
+            .await?;
+        Ok(())
+    }
+}
+
+#[derive(Iden)]
+enum Aggregator {
+    Table,
+    BearerToken,
+    EncryptedBearerToken,
+}
+
+#[derive(Iden)]
+enum Task {
+    Table,
+}

src/bin.rs

@@ -9,8 +9,12 @@ async fn main() {
 
     let config = match Config::from_env() {
         Ok(config) => config,
-        Err(e) => panic!("{e}"),
+        Err(e) => {
+            eprintln!("{e}");
+            std::process::exit(-1);
+        }
     };
+
     let stopper = Stopper::new();
     let observer = CloneCounterObserver::default();
 

src/clients/aggregator_client.rs

@@ -15,9 +15,9 @@ const CONTENT_TYPE: &str = "application/vnd.janus.aggregator+json;version=0.1";
 #[derive(Debug, Clone)]
 pub struct AggregatorClient {
     client: Client,
-    base_url: Url,
     auth_header: HeaderValue,
     aggregator: Aggregator,
+    base_url: Url,
 }
 
 impl AsRef<Client> for AggregatorClient {
@@ -27,17 +27,17 @@ impl AsRef<Client> for AggregatorClient {
 }
 
 impl AggregatorClient {
-    pub fn new(client: Client, aggregator: Aggregator) -> Self {
+    pub fn new(client: Client, aggregator: Aggregator, bearer_token: &str) -> Self {
         let mut base_url: Url = aggregator.api_url.clone().into();
         if !base_url.path().ends_with('/') {
             base_url.set_path(&format!("{}/", base_url.path()));
         }
 
         Self {
             client,
-            base_url,
-            auth_header: HeaderValue::from(format!("Bearer {}", &aggregator.bearer_token)),
+            auth_header: format!("Bearer {bearer_token}").into(),
             aggregator,
+            base_url,
         }
     }
 

src/config.rs

@@ -1,4 +1,4 @@
-use crate::handler::oauth2::Oauth2Config;
+use crate::{handler::oauth2::Oauth2Config, Crypter};
 use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine};
 use email_address::EmailAddress;
 use std::{collections::VecDeque, env::VarError, error::Error, str::FromStr};
@@ -17,6 +17,7 @@ pub struct Config {
     pub auth_client_secret: String,
     pub auth_url: Url,
     pub client: Client,
+    pub crypter: Crypter,
     pub database_url: Url,
     pub email_address: EmailAddress,
     pub postmark_token: String,
@@ -95,6 +96,7 @@ impl Config {
             auth_client_secret: var("AUTH_CLIENT_SECRET")?,
             auth_url: var("AUTH_URL")?,
             client: build_client(),
+            crypter: var("DATABASE_ENCRYPTION_KEYS")?,
             database_url: var("DATABASE_URL")?,
             email_address: var("EMAIL_ADDRESS")?,
             postmark_token: var("POSTMARK_TOKEN")?,

src/crypter.rs

@@ -0,0 +1,142 @@
+use aes_gcm::{
+    aead::{AeadCore, AeadInPlace, KeyInit, OsRng},
+    Aes128Gcm as AesGcm, Error, KeySizeUser, Nonce,
+};
+use base64::{engine::general_purpose::URL_SAFE_NO_PAD, DecodeError, Engine};
+use std::{
+    collections::VecDeque,
+    fmt::{self, Debug, Formatter},
+    iter,
+    str::FromStr,
+    sync::Arc,
+};
+use typenum::marker_traits::Unsigned;
+
+pub type Key = aes_gcm::Key<AesGcm>;
+
+#[derive(Clone)]
+pub struct Crypter(Arc<CrypterInner>);
+
+#[derive(thiserror::Error, Debug, Clone)]
+pub enum CrypterParseError {
+    #[error(transparent)]
+    Base64(#[from] DecodeError),
+
+    #[error("incorrect key length, must be {} bytes after base64 decode", <AesGcm as KeySizeUser>::KeySize::USIZE)]
+    IncorrectLength,
+
+    #[error("at least one key needed")]
+    Missing,
+}
+
+impl FromStr for Crypter {
+    type Err = CrypterParseError;
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let mut keys = s
+            .split(',')
+            .map(|s| {
+                URL_SAFE_NO_PAD
+                    .decode(s)
+                    .map_err(CrypterParseError::Base64)
+                    .and_then(|v| Key::from_exact_iter(v).ok_or(CrypterParseError::IncorrectLength))
+            })
+            .collect::<Result<VecDeque<Key>, _>>()?;
+        let current_key = keys.pop_front().ok_or(CrypterParseError::Missing)?;
+        Ok(Self::new(current_key, keys))
+    }
+}
+
+impl Debug for Crypter {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Crypter")
+            .field("current_ciphers", &"..")
+            .field("past_ciphers", &self.0.past_ciphers.len())
+            .finish()
+    }
+}
+
+#[derive(Clone)]
+struct CrypterInner {
+    current_cipher: AesGcm,
+    past_ciphers: Vec<AesGcm>,
+}
+
+impl From<Key> for Crypter {
+    fn from(key: Key) -> Self {
+        Self::new(key, [])
+    }
+}
+
+impl Crypter {
+    pub fn new(current_key: Key, past_keys: impl IntoIterator<Item = Key>) -> Self {
+        Self(Arc::new(CrypterInner {
+            current_cipher: AesGcm::new(&current_key),
+            past_ciphers: past_keys.into_iter().map(|key| AesGcm::new(&key)).collect(),
+        }))
+    }
+
+    pub fn generate_key() -> Key {
+        AesGcm::generate_key(OsRng)
+    }
+
+    pub fn encrypt(&self, associated_data: &[u8], plaintext: &[u8]) -> Result<Vec<u8>, Error> {
+        self.0.encrypt(associated_data, plaintext)
+    }
+
+    pub fn decrypt(
+        &self,
+        associated_data: &[u8],
+        nonce_and_ciphertext: &[u8],
+    ) -> Result<Vec<u8>, Error> {
+        self.0.decrypt(associated_data, nonce_and_ciphertext)
+    }
+}
+
+impl CrypterInner {
+    fn encrypt(&self, associated_data: &[u8], plaintext: &[u8]) -> Result<Vec<u8>, Error> {
+        let nonce = AesGcm::generate_nonce(&mut OsRng);
+        let mut buffer = plaintext.to_vec();
+        self.current_cipher
+            .encrypt_in_place(&nonce, associated_data, &mut buffer)?;
+        let mut nonce_and_ciphertext = nonce.to_vec();
+        nonce_and_ciphertext.append(&mut buffer);
+        Ok(nonce_and_ciphertext)
+    }
+
+    fn decrypt(
+        &self,
+        associated_data: &[u8],
+        nonce_and_ciphertext: &[u8],
+    ) -> Result<Vec<u8>, Error> {
+        let nonce_size = <AesGcm as AeadCore>::NonceSize::USIZE;
+        if nonce_and_ciphertext.len() < nonce_size {
+            return Err(Error);
+        }
+
+        let (nonce, ciphertext) = nonce_and_ciphertext.split_at(nonce_size);
+
+        self.cipher_iter()
+            .find_map(|cipher| {
+                self.decrypt_with_key(cipher, associated_data, nonce, ciphertext)
+                    .ok()
+            })
+            .ok_or(Error)
+    }
+
+    fn cipher_iter(&self) -> impl Iterator<Item = &AesGcm> {
+        iter::once(&self.current_cipher).chain(self.past_ciphers.iter())
+    }
+
+    fn decrypt_with_key(
+        &self,
+        cipher: &AesGcm,
+        associated_data: &[u8],
+        nonce: &[u8],
+        ciphertext: &[u8],
+    ) -> Result<Vec<u8>, Error> {
+        let nonce = Nonce::from_slice(nonce);
+        let mut bytes = ciphertext.to_vec();
+        cipher.decrypt_in_place(nonce, associated_data, &mut bytes)?;
+        Ok(bytes)
+    }
+}

src/entity/aggregator.rs

@@ -5,12 +5,13 @@ mod update_aggregator;
 mod vdaf_name;
 
 use super::{url::Url, AccountColumn, AccountRelation, Accounts, Memberships};
-use crate::clients::AggregatorClient;
+use crate::{clients::AggregatorClient, Crypter, Error};
 use sea_orm::{
     ActiveModelBehavior, ActiveValue, DeriveEntityModel, DerivePrimaryKey, DeriveRelation,
     EntityTrait, EnumIter, IntoActiveModel, PrimaryKeyTrait, Related, RelationDef, RelationTrait,
 };
 use serde::{Deserialize, Serialize};
+
 use time::OffsetDateTime;
 use uuid::Uuid;
 
@@ -39,10 +40,10 @@ pub struct Model {
     pub dap_url: Url,
     pub api_url: Url,
     pub is_first_party: bool,
-    #[serde(skip)]
-    pub bearer_token: String,
     pub query_types: QueryTypeNameSet,
     pub vdafs: VdafNameSet,
+    #[serde(skip)]
+    pub encrypted_bearer_token: Vec<u8>,
 }
 
 impl Model {
@@ -57,8 +58,24 @@ impl Model {
         self.deleted_at.is_some()
     }
 
-    pub fn client(&self, http_client: trillium_client::Client) -> AggregatorClient {
-        AggregatorClient::new(http_client, self.clone())
+    pub fn client(
+        &self,
+        http_client: trillium_client::Client,
+        crypter: &Crypter,
+    ) -> Result<AggregatorClient, Error> {
+        Ok(AggregatorClient::new(
+            http_client,
+            self.clone(),
+            &self.bearer_token(crypter)?,
+        ))
+    }
+
+    pub fn bearer_token(&self, crypter: &Crypter) -> Result<String, Error> {
+        let bearer_token_bytes = crypter.decrypt(
+            self.api_url.as_ref().as_bytes(),
+            &self.encrypted_bearer_token,
+        )?;
+        String::from_utf8(bearer_token_bytes).map_err(Into::into)
     }
 }