feat: implements allow_new_users setting

docs/src/config.md

@@ -106,6 +106,7 @@ The following configuration options are available.
 | <span id="SYNC_TOKENSERVER__STATSD_LABEL"></span>SYNC_TOKENSERVER__STATSD_LABEL | syncstorage.tokenserver | StatsD metrics label prefix |
 | <span id="SYNC_TOKENSERVER__TOKEN_DURATION"></span>SYNC_TOKENSERVER__TOKEN_DURATION | 3600 | Token TTL (1 hour) |
 | <span id="SYNC_TOKENSERVER__FXA_WEBHOOK_ENABLED"></span>SYNC_TOKENSERVER__FXA_WEBHOOK_ENABLED | false | Enable the FxA webhook endpoint. When disabled, the route is not registered. |
+| <span id="SYNC_TOKENSERVER__ALLOW_NEW_USERS"></span>SYNC_TOKENSERVER__ALLOW_NEW_USERS | true | Whether new users may be created. When disabled, only previously registered users can use the tokenserver service. |
 | <span id="SYNC_TOKENSERVER__FXA_WEBHOOK_METRICS_ONLY"></span>SYNC_TOKENSERVER__FXA_WEBHOOK_METRICS_ONLY | false | Run the FxA webhook handler in metrics-only mode. Received events are counted but not processed. Only used if `FXA_WEBHOOK_ENABLED` is true. |
 | <span id="SYNC_TOKENSERVER__FXA_WEBHOOK_SET_CLIENT_ID"></span>SYNC_TOKENSERVER__FXA_WEBHOOK_SET_CLIENT_ID | None | Expected `aud` of FxA Security Event Tokens. Required for account event webhooks. |
 | <span id="SYNC_TOKENSERVER__FXA_WEBHOOK_SET_ISSUER"></span>SYNC_TOKENSERVER__FXA_WEBHOOK_SET_ISSUER | None | Expected `iss` of FxA Security Event Tokens. Required for account event webhooks. |

syncserver/src/tokenserver/extractors.rs

@@ -249,6 +249,7 @@ impl FromRequest for TokenserverRequest {
                     client_state: auth_data.client_state.clone(),
                     keys_changed_at: auth_data.keys_changed_at,
                     capacity_release_rate: state.node_capacity_release_rate,
+                    allow_new_users: state.allow_new_users,
                 })
                 .await?;
             log_items_mutator.insert("first_seen_at".to_owned(), user.first_seen_at.to_string());
@@ -1345,6 +1346,7 @@ mod tests {
             token_duration: TOKEN_DURATION,
             set_verifiers: Vec::new(),
             fxa_webhook_enabled: false,
+            allow_new_users: true,
             fxa_webhook_metrics_only: false,
         }
     }
@@ -1368,6 +1370,7 @@ mod tests {
             token_duration: TOKEN_DURATION,
             set_verifiers,
             fxa_webhook_enabled: true,
+            allow_new_users: true,
             fxa_webhook_metrics_only: false,
         }
     }

syncserver/src/tokenserver/handlers.rs

@@ -461,6 +461,7 @@ mod tests {
             token_duration: 3600,
             set_verifiers,
             fxa_webhook_enabled: true,
+            allow_new_users: true,
             fxa_webhook_metrics_only: false,
         }
     }

syncserver/src/tokenserver/mod.rs

@@ -34,6 +34,7 @@ pub struct ServerState {
     pub token_duration: u64,
     pub set_verifiers: Vec<SETVerifierImpl>,
     pub fxa_webhook_enabled: bool,
+    pub allow_new_users: bool,
     pub fxa_webhook_metrics_only: bool,
 }
 
@@ -111,6 +112,7 @@ impl ServerState {
             token_duration: settings.token_duration,
             set_verifiers,
             fxa_webhook_enabled: settings.fxa_webhook_enabled,
+            allow_new_users: settings.allow_new_users,
             fxa_webhook_metrics_only: settings.fxa_webhook_metrics_only,
         })
     }

tokenserver-db-common/src/error.rs

@@ -18,6 +18,10 @@ pub struct DbError {
 }
 
 impl DbError {
+    pub fn is_user_not_created(user: String) -> Self {
+        DbErrorKind::UserNotCreated(user).into()
+    }
+
     pub fn internal(msg: String) -> Self {
         DbErrorKind::Internal(msg).into()
     }
@@ -26,6 +30,11 @@ impl DbError {
         DbErrorKind::PoolTimeout(timeout_type).into()
     }
 
+    #[cfg(debug_assertions)]
+    pub fn is_user_not_created(&self) -> bool {
+        matches!(&self.kind, DbErrorKind::UserNotCreated(_))
+    }
+
     #[cfg(debug_assertions)]
     pub fn is_diesel_not_found(&self) -> bool {
         matches!(&self.kind, DbErrorKind::Sql(e) if e.is_diesel_not_found())
@@ -59,6 +68,9 @@ impl ReportableError for DbError {
 
 #[derive(Debug, Error)]
 enum DbErrorKind {
+    #[error("Specified user does not exist (new users are disallowed): {}", _0)]
+    UserNotCreated(String),
+
     #[error("{}", _0)]
     Sql(SqlError),
 
@@ -72,6 +84,11 @@ enum DbErrorKind {
 impl From<DbErrorKind> for DbError {
     fn from(kind: DbErrorKind) -> Self {
         match kind {
+            DbErrorKind::UserNotCreated(_) => Self {
+                kind,
+                status: StatusCode::FORBIDDEN,
+                backtrace: Box::new(Backtrace::new_unresolved()),
+            },
             DbErrorKind::Sql(ref sqle) => Self {
                 status: sqle.status,
                 backtrace: Box::new(sqle.backtrace.clone()),

tokenserver-db-common/src/lib.rs

@@ -113,7 +113,7 @@ pub trait Db {
     fn metrics(&self) -> &Metrics;
 
     /// Gets the user with the given email and service ID.
-    /// If one doesn't exist, allocates a new user.
+    /// If one doesn't exist, and when configuration allows it, allocates a new user.
     async fn get_or_create_user(
         &mut self,
         params: params::GetOrCreateUser,
@@ -127,7 +127,11 @@ pub trait Db {
 
         if raw_users.is_empty() {
             // There are no users in the database with the given email and service ID, so
-            // allocate a new one.
+            // allocate a new one if allowed by configuration.
+            if !params.allow_new_users {
+                return Err(DbError::user_not_created(params.email.clone()));
+            }
+
             let allocate_user_result = self
                 .allocate_user(params.clone() as params::AllocateUser)
                 .await?;
@@ -193,6 +197,7 @@ pub trait Db {
                             client_state: raw_user.client_state.clone(),
                             keys_changed_at: raw_user.keys_changed_at,
                             capacity_release_rate: params.capacity_release_rate,
+                            allow_new_users: params.allow_new_users,
                         })
                         .await?
                     };

tokenserver-db-common/src/params.rs

@@ -35,6 +35,7 @@ pub struct GetOrCreateUser {
     pub client_state: String,
     pub keys_changed_at: Option<i64>,
     pub capacity_release_rate: Option<f32>,
+    pub allow_new_users: bool,
 }
 
 pub type AllocateUser = GetOrCreateUser;

tokenserver-db/src/tests.rs

@@ -556,6 +556,7 @@ async fn test_node_allocation() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
     assert_eq!(user.node, "https://node1");
@@ -609,6 +610,7 @@ async fn test_allocation_to_least_loaded_node() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -620,6 +622,7 @@ async fn test_allocation_to_least_loaded_node() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -663,6 +666,7 @@ async fn test_allocation_is_not_allowed_to_downed_nodes() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await;
     let error = result.unwrap_err();
@@ -704,6 +708,7 @@ async fn test_allocation_is_not_allowed_to_backoff_nodes() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await;
     let error = result.unwrap_err();
@@ -744,6 +749,7 @@ async fn test_node_reassignment_when_records_are_replaced() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
     let user1 = db
@@ -768,6 +774,7 @@ async fn test_node_reassignment_when_records_are_replaced() -> DbResult<()> {
             client_state: "bbbb".to_owned(),
             keys_changed_at: Some(1235),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -816,6 +823,7 @@ async fn test_node_reassignment_not_done_for_retired_users() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -827,6 +835,7 @@ async fn test_node_reassignment_not_done_for_retired_users() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -884,6 +893,7 @@ async fn test_node_reassignment_and_removal() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -895,6 +905,7 @@ async fn test_node_reassignment_and_removal() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -906,6 +917,7 @@ async fn test_node_reassignment_and_removal() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -917,6 +929,7 @@ async fn test_node_reassignment_and_removal() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -949,6 +962,7 @@ async fn test_node_reassignment_and_removal() -> DbResult<()> {
                 client_state: user.client_state.clone(),
                 keys_changed_at: user.keys_changed_at,
                 capacity_release_rate: None,
+                allow_new_users: true,
             })
             .await?;
 
@@ -978,6 +992,7 @@ async fn test_node_reassignment_and_removal() -> DbResult<()> {
                 client_state: user.client_state.clone(),
                 keys_changed_at: user.keys_changed_at,
                 capacity_release_rate: None,
+                allow_new_users: true,
             })
             .await?;
 
@@ -1034,6 +1049,7 @@ async fn test_gradual_release_of_node_capacity() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -1051,6 +1067,7 @@ async fn test_gradual_release_of_node_capacity() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -1070,6 +1087,7 @@ async fn test_gradual_release_of_node_capacity() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -1087,6 +1105,7 @@ async fn test_gradual_release_of_node_capacity() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -1105,6 +1124,7 @@ async fn test_gradual_release_of_node_capacity() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -1122,6 +1142,7 @@ async fn test_gradual_release_of_node_capacity() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -1140,6 +1161,7 @@ async fn test_gradual_release_of_node_capacity() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await;
 
@@ -1185,6 +1207,7 @@ async fn test_correct_created_at_used_during_node_reassignment() -> DbResult<()>
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -1204,6 +1227,7 @@ async fn test_correct_created_at_used_during_node_reassignment() -> DbResult<()>
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -1245,6 +1269,7 @@ async fn test_correct_created_at_used_during_user_retrieval() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -1261,6 +1286,7 @@ async fn test_correct_created_at_used_during_user_retrieval() -> DbResult<()> {
             client_state: "aaaa".to_owned(),
             keys_changed_at: Some(1234),
             capacity_release_rate: None,
+            allow_new_users: true,
         })
         .await?;
 
@@ -1270,6 +1296,46 @@ async fn test_correct_created_at_used_during_user_retrieval() -> DbResult<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn test_no_new_user_allocation() -> DbResult<()> {
+    let pool = db_pool().await?;
+    let mut db = pool.get().await?;
+
+    let service_id = db
+        .get_service_id(params::GetServiceId {
+            service: "sync-1.5".to_owned(),
+        })
+        .await?
+        .id;
+
+    // Add a node
+    db.post_node(params::PostNode {
+        service_id,
+        node: "https://node1".to_owned(),
+        current_load: 4,
+        capacity: 8,
+        available: 1,
+        ..Default::default()
+    })
+    .await?;
+
+    // Try to create a user
+    let result = db
+        .get_or_create_user(params::GetOrCreateUser {
+            service_id,
+            generation: 1234,
+            email: "test4@test.com".to_owned(),
+            client_state: "aaaa".to_owned(),
+            keys_changed_at: Some(1234),
+            capacity_release_rate: None,
+            allow_new_users: false,
+        })
+        .await;
+    assert!(result.unwrap_err().is_user_not_created());
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_latest_created_at() -> DbResult<()> {
     let pool = db_pool().await?;