feat(auth): Add support for multiple api keys (supabase#307)

Author: iambriccardo

etl-api/README.md

@@ -23,6 +23,7 @@ This API service provides a RESTful interface for managing Postgres replication
 - [Prerequisites](#prerequisites)
 - [Development](#development)
 - [Environment Variables](#environment-variables)
+- [Authentication](#authentication)
 
 ## Prerequisites
 
@@ -80,3 +81,16 @@ After making changes to the database schema, update the SQLx metadata:
 ```bash
 cargo sqlx prepare
 ```
+
+## Authentication
+
+- The API uses Bearer token auth via the `Authorization` header.
+- Configure authentication with `api_keys` (each is base64 of 32 random bytes). All listed keys are accepted, enabling seamless key rotation.
+
+Config example (YAML):
+
+```yaml
+api_keys:
+  - XOUbHmWbt9h7nWl15wWwyWQnctmFGNjpawMc3lT5CFs=
+  - h1QqT7u+8t4q0t3m8rjOa2qK7F8w6h9C1xYzPqL7pmc=
+```

etl-api/configuration/dev.yaml

@@ -14,4 +14,6 @@ application:
 encryption_key:
   id: 0
   key: BlK9AlrzqRnCZy53j42uE1p2qGBiF7HYZjZYFaZObqg=
-api_key: XOUbHmWbt9h7nWl15wWwyWQnctmFGNjpawMc3lT5CFs=
+api_keys:
+  - XOUbHmWbt9h7nWl15wWwyWQnctmFGNjpawMc3lT5CFs=
+  - jD3rU2aYq7nVp1mWb4sTxQ9PZkHcGdR8eM5oLfNhJ0s=

etl-api/src/authentication.rs

@@ -21,29 +21,43 @@ pub async fn auth_validator(
         .unwrap_or_default()
         .scope("v1");
 
-    let api_key = req
+    let api_config = req
         .app_data::<Data<ApiConfig>>()
-        .expect("missing api configuration")
-        .api_key
-        .as_str();
+        .expect("Missing API configuration while doing authentication");
 
     let token = credentials.token();
-
-    let api_key: ApiKey = match api_key.try_into() {
-        Ok(api_key) => api_key,
+    let token: ApiKey = match token.try_into() {
+        Ok(token) => token,
         Err(_) => {
             return Err((AuthenticationError::from(config).into(), req));
         }
     };
 
-    let token: ApiKey = match token.try_into() {
-        Ok(token) => token,
-        Err(_) => {
+    // Decode all configured API keys (rotation supported via multiple entries).
+    let configured_keys: Vec<ApiKey> = {
+        let keys = &api_config.api_keys;
+        if keys.is_empty() {
             return Err((AuthenticationError::from(config).into(), req));
         }
+
+        let mut configured_keys = Vec::with_capacity(keys.len());
+        for key in keys {
+            match key.as_str().try_into() {
+                Ok(k) => configured_keys.push(k),
+                Err(_) => return Err((AuthenticationError::from(config).into(), req)),
+            }
+        }
+
+        configured_keys
     };
 
-    if !constant_time_eq_n(&api_key.key, &token.key) {
+    // Compare against all configured keys without an early exit to avoid timing leaks.
+    let mut valid = false;
+    for key in &configured_keys {
+        valid |= constant_time_eq_n(&key.key, &token.key);
+    }
+
+    if !valid {
         return Err((AuthenticationError::from(config).into(), req));
     }
 

etl-api/src/config.rs

@@ -1,4 +1,5 @@
 use base64::{Engine, prelude::BASE64_STANDARD};
+use etl_config::Config;
 use etl_config::shared::{PgConnectionConfig, SentryConfig};
 use serde::de::{MapAccess, Visitor};
 use serde::{Deserialize, Deserializer, de};
@@ -20,12 +21,18 @@ pub struct ApiConfig {
     pub application: ApplicationSettings,
     /// Encryption key configuration.
     pub encryption_key: EncryptionKey,
-    /// Base64-encoded API key string.
-    pub api_key: String,
+    /// List of base64-encoded API keys.
+    ///
+    /// All keys in this list are considered valid for authentication.
+    pub api_keys: Vec<String>,
     /// Optional Sentry configuration for error tracking.
     pub sentry: Option<SentryConfig>,
 }
 
+impl Config for ApiConfig {
+    const LIST_PARSE_KEYS: &'static [&'static str] = &["api_keys"];
+}
+
 /// HTTP server configuration settings.
 #[derive(Debug, Clone, Deserialize)]
 pub struct ApplicationSettings {

etl-api/tests/common/test_app.rs

@@ -21,6 +21,7 @@ use etl_api::{
 use etl_config::shared::PgConnectionConfig;
 use etl_config::{Environment, load_config};
 use etl_postgres::sqlx::test_utils::drop_pg_database;
+use rand::random_range;
 use reqwest::{IntoUrl, RequestBuilder};
 use std::io;
 use std::net::TcpListener;
@@ -509,7 +510,11 @@ pub async fn spawn_test_app() -> TestApp {
 
     let key = generate_random_key::<32>().expect("failed to generate random key");
     let encryption_key = encryption::EncryptionKey { id: 0, key };
-    let api_key = "XOUbHmWbt9h7nWl15wWwyWQnctmFGNjpawMc3lT5CFs=".to_string();
+
+    // We choose a random API key from the ones configured to show that rotation works.
+    let api_key_index = random_range(0..config.api_keys.len());
+    let api_key = config.api_keys[api_key_index].clone();
+
     let k8s_client = Some(Arc::new(MockK8sClient) as Arc<dyn K8sClient>);
 
     let server = run(

etl-api/tests/integration/mod.rs

@@ -3,9 +3,8 @@ mod destinations_pipelines_test;
 mod etl_tables_missing_test;
 mod health_check_test;
 mod images_test;
+mod metrics_test;
 mod pipelines_test;
 mod sources_test;
 mod tenants_sources_test;
 mod tenants_test;
-
-mod metrics_test;

etl-config/src/load.rs

@@ -19,21 +19,34 @@ const ENV_PREFIX_SEPARATOR: &str = "_";
 /// Example: `APP_DATABASE__URL` sets the `database.url` field.
 const ENV_SEPARATOR: &str = "__";
 
+/// Separator for list elements in environment variables.
+///
+/// Example: `APP_API_KEYS=abc,def` sets the `api_keys` array field.
+const LIST_SEPARATOR: &str = ",";
+
+/// Trait defining the list of keys that should be parsed as lists in a given [`Config`]
+/// implementation.
+pub trait Config {
+    /// Slice containing all the keys that should be parsed as lists when loading the configuration.
+    const LIST_PARSE_KEYS: &'static [&'static str];
+}
+
 /// Loads hierarchical configuration from YAML files and environment variables.
 ///
 /// Loads configuration in this order:
 /// 1. Base configuration from `configuration/base.yaml`
 /// 2. Environment-specific file from `configuration/{environment}.yaml`
 /// 3. Environment variable overrides prefixed with `APP`
 ///
-/// Nested keys use double underscores: `APP_DATABASE__URL` → `database.url`
+/// Nested keys use double underscores: `APP_DATABASE__URL` → `database.url` and lists are separated
+/// by `,`.
 ///
 /// # Panics
 /// Panics if current directory cannot be determined or if `APP_ENVIRONMENT`
 /// cannot be parsed.
 pub fn load_config<T>() -> Result<T, config::ConfigError>
 where
-    T: DeserializeOwned,
+    T: Config + DeserializeOwned,
 {
     let base_path = std::env::current_dir().expect("Failed to determine the current directory");
     let configuration_directory = base_path.join(CONFIGURATION_DIR);
@@ -43,21 +56,33 @@ where
     let environment = Environment::load().expect("Failed to parse APP_ENVIRONMENT.");
 
     let environment_filename = format!("{environment}.yaml");
+
+    // We build the environment configuration source.
+    let mut environment_source = config::Environment::with_prefix(ENV_PREFIX)
+        .prefix_separator(ENV_PREFIX_SEPARATOR)
+        .separator(ENV_SEPARATOR)
+        .try_parsing(true)
+        .list_separator(LIST_SEPARATOR);
+
+    // For all the list parse keys, we add them to the environment source. These are used to define
+    // which keys should be parsed as lists.
+    for key in <T as Config>::LIST_PARSE_KEYS {
+        environment_source = environment_source.with_list_parse_key(key);
+    }
+
     let settings = config::Config::builder()
+        // Add in settings from the base configuration file.
         .add_source(config::File::from(
             configuration_directory.join(BASE_CONFIG_FILE),
         ))
+        // Add in settings from the environment-specific file.
         .add_source(config::File::from(
             configuration_directory.join(environment_filename),
         ))
         // Add in settings from environment variables (with a prefix of APP and '__' as separator)
         // E.g. `APP_DESTINATION__BIG_QUERY__PROJECT_ID=my-project-id` sets
         // `Settings { destination: BigQuery { project_id } }` to `my-project-id`.
-        .add_source(
-            config::Environment::with_prefix(ENV_PREFIX)
-                .prefix_separator(ENV_PREFIX_SEPARATOR)
-                .separator(ENV_SEPARATOR),
-        )
+        .add_source(environment_source)
         .build()?;
 
     settings.try_deserialize::<T>()

etl-config/src/shared/connection.rs

@@ -3,8 +3,8 @@ use serde::{Deserialize, Serialize};
 use sqlx::postgres::{PgConnectOptions as SqlxConnectOptions, PgSslMode as SqlxSslMode};
 use tokio_postgres::{Config as TokioPgConnectOptions, config::SslMode as TokioPgSslMode};
 
-use crate::SerializableSecretString;
 use crate::shared::ValidationError;
+use crate::{Config, SerializableSecretString};
 
 /// Postgres server options for ETL workloads.
 ///
@@ -133,6 +133,10 @@ pub struct PgConnectionConfig {
     pub tls: TlsConfig,
 }
 
+impl Config for PgConnectionConfig {
+    const LIST_PARSE_KEYS: &'static [&'static str] = &[];
+}
+
 /// TLS configuration for secure Postgres connections.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct TlsConfig {

etl-config/src/shared/replicator.rs

@@ -1,3 +1,4 @@
+use crate::Config;
 use crate::shared::pipeline::PipelineConfig;
 use crate::shared::{DestinationConfig, SentryConfig, SupabaseConfig, ValidationError};
 use serde::{Deserialize, Serialize};
@@ -33,3 +34,7 @@ impl ReplicatorConfig {
         self.pipeline.validate()
     }
 }
+
+impl Config for ReplicatorConfig {
+    const LIST_PARSE_KEYS: &'static [&'static str] = &[];
+}